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

[likebean]

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 Messages (ADR-0019)	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_snapshots 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?

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/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/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: Python: [Feature]: AG-UI runs should survive client disconnect (detached execution + resumable event stream) #7230 (disconnect vs snapshot save), Python: AG-UI snapshot replay with service sessions blocks Foundry Hosted Agents #7710 (snapshot replay vs service sessions)

Code Sample

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

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):

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