Python: [Feature]: Compose AG-UI snapshots with HistoryProvider, pagination, and compaction

### Description

## Summary

Please publish an **official architecture + complete demo** for long-running AG-UI chats that shows how these pieces are supposed to work **together**:

1. **Where conversation history is persisted** — AG-UI Thread Snapshot store vs Agent `HistoryProvider`
2. **How a long transcript is loaded for the UI** (pagination / incremental hydrate)
3. **What is actually sent to `agent.run`** each turn (full reconstruct vs new turn only)
4. **How MAF compaction is applied without re-compacting the whole history every request**

Today the components exist, but AG-UI demos and the Python AG-UI runner do not compose them. Building a production chat on `add_agent_framework_fastapi_endpoint` forces apps to invent a fourth persistence/compaction path, or to re-compact the full snapshot on every turn.

This is a design/docs/sample request, not a runtime bug report.

## Why this feels disconnected

MAF already has three well-designed subsystems:

| Subsystem | Designed for | Official samples |
|---|---|---|
| AG-UI Thread Snapshot (`AGUIThreadSnapshotStore`) | Stateless HTTP, refresh/hydrate, HITL interrupt replay | `python/samples/05-end-to-end/ag_ui_assistant_ui_chat`, `ag_ui_single_agent` |
| `HistoryProvider` + `AgentSession` | Same in-process session, `agent.run("new turn", session=session)` | `python/samples/02-agents/conversations`, `compaction/compaction_provider.py` |
| Compaction (`CompactionStrategy`, in-run client compaction, `CompactionProvider`) | Token window + persist `_excluded`/summaries on stored `Message`s ([ADR-0019](https://github.com/microsoft/agent-framework/blob/main/docs/decisions/0019-python-context-compaction-strategy.md)) | `python/samples/02-agents/compaction/` |

The AG-UI Python runner currently makes snapshot.messages the **conversation-history authority** (`python/packages/ag-ui/README.md`) and **excludes `HistoryProvider` keys from `session_state`**. Each follow-up turn `_reconstruct_messages_from_thread_snapshot`s the stored UI transcript and passes that **full list** as `input_messages` to `agent.run(...)`.

Consequences:

- `InMemoryHistoryProvider` auto-injected for a local session does not survive the next AG-UI request (provider state is stripped from the snapshot).
- `CompactionProvider.before_run` compactes **provider-loaded context**, not AG-UI `input_messages`, so it never sees the reconstructed transcript.
- Setting `Agent(..., compaction_strategy=...)` *does* compact for that model call (MAF in-run path), but the next request reconstructs the **uncompacted** snapshot again, so summarization would be repeated every turn.
- Compacting inside an app-owned snapshot store uses `apply_compaction` as a standalone function, which is **not** the Agent/CompactionProvider pipeline.

So the question is not “does MAF have compaction?” — it does. It is: **what is the supported composition with AG-UI?**

```text
Today (AG-UI demo path)
Client --(threadId + latest user message)--> endpoint
reconstruct full snapshot.messages
agent.run(full history) # HistoryProvider / CompactionProvider mostly unused
save full UI transcript to snapshot

Needed for long chats
Snapshot store -> UI hydrate / scroll / HITL replay
HistoryProvider -> model context (windowed / compacted, with annotations)
CompactionProvider.after_run persists exclusions; next before_run skip_excluded
agent.run(new turn only, session restored from session_state)
```

## There is no config switch for “new turn only”

We looked for an existing flag before asking for a sample. There is none.

`AgentConfig` / `add_agent_framework_fastapi_endpoint` expose `snapshot_store`, `use_service_session`, `require_confirmation`, `state_schema`, and `predict_state_config`. None of these change `agent.run` to receive only the new user turn while `HistoryProvider` loads prior context.

- `snapshot_store` enables hydrate + reconstruct of the **full** UI transcript into `agent.run`.
- `use_service_session` only sets `AgentSession.service_session_id` (hosted conversation id). Local `messages` are still the reconstructed list.
- `HistoryProvider(load_messages=False)` only skips provider load inside `Agent`; AG-UI still passes the full reconstruct as `input_messages`.

The exclusion of HistoryProvider state from snapshot continuation is **hardcoded**, not optional:

```python
# python/packages/ag-ui/agent_framework_ag_ui/_agent_run.py
# _serialize_session_continuation_state
excluded_keys = {
*shared_state_keys,
_TOOL_APPROVAL_STATE_KEY,
*(provider.source_id for provider in context_providers if isinstance(provider, HistoryProvider)),
}
```

This is covered by an explicit test whose docstring is the current contract:

```python
# python/packages/ag-ui/tests/ag_ui/test_endpoint.py
# test_endpoint_excludes_history_provider_state_from_continuation
"""Snapshot messages remain the sole conversation-history authority."""
```

That test attaches `InMemoryHistoryProvider` to the agent and still asserts turn 2 is the **full reconstructed list** (`First` + `Reply 1` + `Second`), not just `Second`.

So “attach a HistoryProvider and AG-UI will start sending deltas” is not something an app can turn on today. If pattern B below is the intended long-chat design, it is a runner/`_agent_run.py` behavior change (and that test’s contract would need to change), not a missed configuration knob.

## Concrete questions for maintainers

### 1. Persistence: snapshot store vs history store

For an AG-UI FastAPI endpoint, what is the **supported** split?

- Snapshot store = UI-replayable transcript + shared state + interrupts
- HistoryProvider = model context (possibly compacted), persisted via `AGUIThreadSnapshot.session_state` (or an external Cosmos/Redis provider)

Or is snapshot.messages intentionally the only history, and HistoryProvider should stay opt-in/off for AG-UI?

The README “State Authority” table currently says conversation history belongs to the thread snapshot. That conflicts with ADR-0019’s compaction persistence model (annotations live on HistoryProvider messages). Please confirm the intended authority, including for HITL / tool-approval resume.

### 2. Pagination for a long UI transcript

Hydrate today is “empty `messages` + `threadId` → replay the **latest full** `MESSAGES_SNAPSHOT`”. For a multi-thousand-message thread that is not viable.

Is pagination in scope for AG-UI + MAF?

- Cursor/window APIs on the snapshot store for the **frontend** (sidebar, scroll-back, hydrate last N)
- Explicitly **not** pagination into `agent.run` (tool-call groups must stay intact)

If pagination is application-owned, please say so and show it in a sample (`GET /threads/{id}/messages?cursor=`).

### 3. Is feeding the agent the full reconstructed history every turn the recommended design?

Short chats: yes, simplest.

Long chats: token cost, latency, and it bypasses HistoryProvider. Is the official long-chat pattern:

- **A.** Keep reconstructing the full snapshot into `agent.run` (compaction only as a per-call projection; accept re-work or compact the snapshot itself), or
- **B.** When the agent has a `HistoryProvider`, send **only the new user turn** (plus resume payloads), restore `session_state` including history, let `skip_excluded=True` load the compacted projection?

Please pick A or B (or C: service-managed `use_service_session`) as the documented AG-UI default for long threads.

If B: please also say whether that should be automatic when a `HistoryProvider` is present, or a new explicit endpoint/`AgentConfig` flag (today neither exists — see “There is no config switch” above).

### 4. Compaction without re-compacting every request

Desired behavior matches `compaction_provider.py`:

- Persist `_excluded` / summary messages on stored history
- Next turn: strategies no-op under threshold; `SummarizationStrategy` does not re-summarize already-replaced groups
- In-run `compaction_strategy` still runs inside one `agent.run` tool loop

That requires the **next AG-UI request to restore the annotated history**, not a clean UI transcript. Today that round-trip is blocked by `_serialize_session_continuation_state` excluding HistoryProvider keys.

If B above is correct, the glue change is in `agent-framework-ag-ui` (`_agent_run.py` reconstruct + session_state exclusion), not in core compaction. Please confirm so apps do not implement a parallel snapshot-store compactor.

## Requested deliverable

An **official end-to-end sample** (Python AG-UI, ideally extending `ag_ui_assistant_ui_chat`) that demonstrates, in one app:

1. Durable snapshot store (already in that sample)
2. UI-side history pagination or “hydrate last N + load older”
3. Agent-side history via `HistoryProvider` (in-memory for the sample, with a comment for Cosmos/Redis)
4. `CompactionProvider` + a cheap strategy (e.g. `SlidingWindowStrategy`) so compaction **persists across AG-UI turns**
5. A README architecture section: what is stored where, what is sent to the model, what the UI hydrates

Even a short “AG-UI + context engineering” doc that answers the four questions would unblock implementers. A sample is much better: the current packages work in isolation, and the missing piece is how they are wired at the AG-UI boundary.

## Related

- ADR-0019: `docs/decisions/0019-python-context-compaction-strategy.md`
- AG-UI snapshots: `python/packages/ag-ui/README.md` (“AG-UI Thread Snapshots”, State Authority table)
- Reconstruct: `python/packages/ag-ui/agent_framework_ag_ui/_run_common.py` (`_reconstruct_messages_from_thread_snapshot`)
- HistoryProvider stripped from snapshot: `python/packages/ag-ui/agent_framework_ag_ui/_agent_run.py` (`_serialize_session_continuation_state`)
- Contract test: `python/packages/ag-ui/tests/ag_ui/test_endpoint.py` (`test_endpoint_excludes_history_provider_state_from_continuation`)
- Compaction sample that does *not* use AG-UI: `python/samples/02-agents/compaction/compaction_provider.py`
- Nearby AG-UI persistence issues: #7230 (disconnect vs snapshot save), #7710 (snapshot replay vs service sessions)

### Code Sample

Current AG-UI demo path (no HistoryProvider, no compaction, full reconstruct):

```python
agent = Agent(client=client, instructions="...") # no context_providers, no compaction_strategy

add_agent_framework_fastapi_endpoint(
app,
agent,
path="/agent",
snapshot_store=SqliteAGUIThreadSnapshotStore(...),
snapshot_scope_resolver=lambda _request: "demo",
)
```

What we would like the official long-chat sample to look like (sketch — please correct this if it is the wrong composition):

```python
history = InMemoryHistoryProvider(skip_excluded=True)
compaction = CompactionProvider(
after_strategy=SlidingWindowStrategy(keep_last_groups=20),
)
agent = Agent(
client=client,
instructions="...",
context_providers=[history, compaction],
)

add_agent_framework_fastapi_endpoint(
app,
agent,
path="/agent",
snapshot_store=snapshot_store, # UI transcript + session_state that round-trips history
snapshot_scope_resolver=resolve_scope,
)
# Follow-up POSTs send threadId + the new user message (or hydrate with messages=[]).
# Agent.run receives the new turn; HistoryProvider loads the compacted projection.
```

### Language/SDK

Python