Python: [Feature]: expose harness session state (todos, mode) as AG-UI shared state out of the box

### Description

## 1. Summary

The harness (`create_harness_agent`) ships `TodoProvider` and `AgentModeProvider` **enabled by default**, and both persist their state into the `AgentSession` (`session.state["todo"]`, mode provider's `source_id` bucket). AG-UI clients (CopilotKit and friends) have a first-class shared-state channel (`state_schema`, `StateSnapshotEvent`, `useAgent().state`) that is the natural place to render exactly this data — a live plan/todo panel and a plan/execute mode indicator.

Yet nothing in the framework connects the two. Every `StateSnapshotEvent` is built from `FlowState.current_state`, which is fed only by the stored thread snapshot, the client's `RunAgentInput.state`, `default_state`/schema defaults, `predict_state_config` argument mirroring, and explicit `state_update(...)` markers returned by tools. No code path reads `session.state`. The result: a harness agent hosted over AG-UI maintains a rich, authoritative todo list server-side, and the web client's `state.todos` stays permanently empty unless the application writes a custom bridge.

```mermaid
flowchart LR
subgraph SessionWorld["Harness session state (server-only)"]
TP["TodoProvider
session.state['todo']"]
MP["AgentModeProvider
session.state[mode source_id]"]
end
subgraph AGUIWorld["AG-UI shared state (client-visible)"]
FS["FlowState.current_state"]
SSE["StateSnapshotEvent -> useAgent().state"]
end
TP -. "no shipped bridge" .-> FS
MP -. "no shipped bridge" .-> FS
SU["tool returns state_update(...)"] --> FS
PS["predict_state_config
(streamed tool args)"] --> FS
FS --> SSE
```

## 2. Why the existing mechanisms don't cover this

Both official server→client state mechanisms assume the **application authors the tools**:

1. **`state_update()`** (`agent_framework_ag_ui/_state.py`; re-exported from `agent_framework.ag_ui`) — a tool returns `state_update(text=..., state={...})`, and the endpoint merges the state into `FlowState.current_state` and emits a deterministic `StateSnapshotEvent` after the tool result. This is the pattern the shared-state sample demonstrates, and it works well — but the harness's `todos_add` / `todos_complete` / mode tools are framework-owned, return plain strings, and cannot be edited by the application.
2. **`predict_state_config`** — declaratively mirrors streaming tool-call *arguments* into state. For todos this reflects what the model *asked for*, not the provider's authoritative post-execution list (no ids, no completion status, no reconciliation with items added in earlier turns), so it is a poor substitute for the store contents.

Meanwhile the harness side is deliberately transport-agnostic: `agent_framework/_harness/` contains no references to `state_update` or anything AG-UI. The two features were built independently; the pieces exist on both sides but have never been wired together.

A subtlety that makes an app-side bridge harder than it looks: the AG-UI host builds a **fresh `AgentSession` per request**, and clients want *live* updates while the agent edits its plan mid-run. So a correct bridge must observe session state **at function-invocation time** (after each `todos_*` / mode tool executes), not merely snapshot it at the end of the run.

## 3. Current application-level workaround

We ship a small `FunctionMiddleware` that retrofits the official `state_update` mechanism onto the framework-owned tools:

```python
class SharedStateMiddleware(FunctionMiddleware):
"""Push a {mode, todos} snapshot whenever a harness state tool runs."""

async def process(self, context, call_next):
await call_next()
name = getattr(getattr(context, "function", None), "name", "") or ""
if context.session is None or not name.startswith(("todos_", "mode_")):
return
items = await TodoSessionStore().load_items(context.session, source_id=DEFAULT_TODO_SOURCE_ID)
snapshot = {
"mode": get_agent_mode(context.session),
"todos": [item.to_dict() for item in items],
}
# Re-wrap the tool's text result with the shared-state marker; the AG-UI
# endpoint converts it into a StateSnapshotEvent. Must be list[Content]:
# Content.from_function_result only preserves additional_properties (where
# the marker rides) for a list of Content — a bare Content is re-stringified
# and the marker is lost.
context.result = [state_update(text=_result_text(context.result), state=snapshot)]
```

This works, but every application hosting a harness agent over AG-UI has to rediscover the gap, the `TodoSessionStore` internals, and the `list[Content]` marker-preservation gotcha independently.

## 4. Proposed feature

Any of the following (ordered by our preference); all reuse existing public machinery and change nothing on the wire:

### Option A — shipped middleware in `agent-framework-ag-ui`

Provide the middleware above as a supported class, e.g.:

```python
from agent_framework.ag_ui import HarnessStateSyncMiddleware

agent = create_harness_agent(..., middleware=[HarnessStateSyncMiddleware()])
```

Configurable with the state keys to publish (`todos`, `mode`) and the tool-name prefixes to watch. Smallest surface, opt-in, no changes to the harness or the endpoint.

### Option B — declarative session-state mapping on the endpoint

Extend `add_agent_framework_fastapi_endpoint` with a mapping from shared-state keys to session readers, evaluated after each function invocation (and once at run end):

```python
add_agent_framework_fastapi_endpoint(
app, agent,
state_schema={...},
session_state_sync={
"todos": lambda session: [i.to_dict() for i in ...load_items(session)],
"mode": get_agent_mode,
},
)
```

More general (works for any `ContextProvider`, not just todo/mode), but touches the endpoint's run loop.

### Option C — harness todo/mode tools emit `state_update` natively when hosted over AG-UI

The harness tools could return `state_update(...)`-wrapped results when an AG-UI emitter is present. Cleanest for users (zero wiring) but couples the harness to AG-UI, which the current layering deliberately avoids — mentioned for completeness, not recommended.

## 5. Related issues (checked 2026-07-05)

| Issue | State | Relationship |
|---|---|---|
| [#4177](https://github.com/microsoft/agent-framework/issues/4177) | open | **.NET twin of this request**: MAF has the infrastructure (`AgentSessionStateBag`, `StateSnapshotEvent`/`StateDeltaEvent`) "but lacks the integration layer" — every developer re-writes the same custom middleware. This report is the Python counterpart with the harness todo/mode providers as the concrete case |
| [#3167](https://github.com/microsoft/agent-framework/issues/3167) | closed | Python: produced `state_update()` for application-authored tools — the mechanism Option A builds on |
| [#5197](https://github.com/microsoft/agent-framework/issues/5197) | open | Python: the inbound mirror of this request — client/request state should reach `ContextProvider`s; here, provider state should reach AG-UI clients. Together they describe the missing two-way bridge |
| [#6920](https://github.com/microsoft/agent-framework/issues/6920) | open | the *survival* half of the session/AG-UI disconnect: harness session state also does not persist across AG-UI runs |

## 6. Environment

- `agent-framework-core` 1.10.0, `agent-framework-ag-ui` 1.0.0rc7, Python 3.12, Windows 11
- Frontend: CopilotKit `useAgent()` over the FastAPI AG-UI endpoint
- Repro/absence is trivially observable: host any default `create_harness_agent` agent over AG-UI with `state_schema={"todos": {...}}`, ask for a multi-step plan, and watch `STATE_SNAPSHOT` events — `todos` never changes while the chat shows the plan being built.

### Code Sample

```markdown
from agent_framework.ag_ui import HarnessStateSyncMiddleware

agent = create_harness_agent(..., middleware=[HarnessStateSyncMiddleware()])
```

### Language/SDK

Both