ag-ui-protocol / ag-ui-protocol/ag-ui

flushState in compactEvents starts each batch from {}, breaking multi-batch STATE_DELTA streams

Open
#1,720 0 comments 0 reactions 0 assignees View on GitHub
bug SDK
Dominant language
Python
Stars
15.9k
Forks
1.4k
Avg merge
1d 17h
Merged PRs (30d)
163

Description

## Summary

`flushState` in [`sdks/typescript/packages/client/src/compact/compact.ts`](https://github.com/ag-ui-protocol/ag-ui/blob/main/sdks/typescript/packages/client/src/compact/compact.ts) starts every batch from `state = {}`. When a single run's state events span multiple `flushState` calls (e.g. high-rate STATE_DELTAs split across batch boundaries by upstream batching, or a stream that emits state events outside the bracketed RUN_STARTED / RUN_FINISHED window), the second-and-later batches discard the prior batch's accumulated state. JSON-Patch ops that reference paths the prior batch had populated then fail with `OPERATION_PATH_CANNOT_ADD` / `JsonPointerException`.

## Affected versions

- `@ag-ui/client@0.0.53` (current `latest` on npm; same logic in earlier versions)

## Reproduction

The bug is visible whenever any agent backend emits STATE_SNAPSHOT once, then a high-volume STATE_DELTA stream (e.g. per-token streaming with state mirror updates), and the consolidator's batches don't all align with the snapshot.

Minimal in-test repro:

```ts
import { compactEvents } from "@ag-ui/client";
import { EventType } from "@ag-ui/core";

const events = [
// Run 1 — establishes /run_log
{ type: EventType.RUN_STARTED, threadId: "t", runId: "r1" },
{ type: EventType.STATE_SNAPSHOT, snapshot: { run_log: [] } },
{ type: EventType.STATE_DELTA, delta: [{ op: "add", path: "/run_log/-", value: { kind: "token", text: "hi" } }] },
{ type: EventType.RUN_FINISHED, threadId: "t", runId: "r1" },
// Run 2 — no snapshot; assumes /run_log carries over from run 1
{ type: EventType.RUN_STARTED, threadId: "t", runId: "r2" },
{ type: EventType.STATE_DELTA, delta: [{ op: "add", path: "/run_log/-", value: { kind: "token", text: "there" } }] },
{ type: EventType.RUN_FINISHED, threadId: "t", runId: "r2" },
];

compactEvents(events);
// → throws: OPERATION_PATH_CANNOT_ADD: /run_log/- (run 2's flushState starts from {})
```

In the field, we observed this with `@copilotkit/runtime` (uses `@ag-ui/client`'s consolidator) at ~99 events per turn from a per-token-streaming agent: the STATE_SNAPSHOT landed in batch 1, and run-tree-update STATE_DELTAs spilled into batches 2+, surfacing as a `PatchError: OPERATION_PATH_CANNOT_ADD` in the Next.js dev-server log and a blank UI pane downstream. Workaround we shipped server-side: every STATE_DELTA op is now self-contained (`add /run_log = [full array]` instead of `add /run_log/- = entry`, `add /` instead of `replace /`), at `O(N^2)` wire-cost per turn for run_log.

## Root cause

[`flushState`](https://github.com/ag-ui-protocol/ag-ui/blob/main/sdks/typescript/packages/client/src/compact/compact.ts) starts from an empty object every call:

```ts
function flushState(
stateEvents: (StateSnapshotEvent | StateDeltaEvent)[],
compacted: BaseEvent[],
): void {
if (stateEvents.length === 0) return;
let state: any = {}; // <-- always {}, no prior-batch seed
for (const event of stateEvents) {
if (event.type === EventType.STATE_SNAPSHOT) {
state = structuredClone_(event.snapshot);
} else {
const result = jsonpatch.applyPatch(state, structuredClone_(event.delta), true, false);
state = result.newDocument;
}
}
compacted.push({ type: EventType.STATE_SNAPSHOT, snapshot: state });
}
```

Per JSON Patch (RFC 6902 §A.12), `add` to an array via `/-` requires the array to exist; `replace` on a missing path is an error. The empty-doc start means any delta referencing a path established by a prior batch's snapshot fails.

## Proposed fix

Thread a running state across `flushState` calls within a single `compactEvents` invocation. Backwards-compatible (default `initialState = {}` matches current behaviour):

```ts
function flushState(
stateEvents: (StateSnapshotEvent | StateDeltaEvent)[],
compacted: BaseEvent[],
initialState: any = {}, // new param
): any { // returns the new state for the caller
if (stateEvents.length === 0) return initialState;
let state: any = structuredClone_(initialState);
for (const event of stateEvents) {
if (event.type === EventType.STATE_SNAPSHOT) {
state = structuredClone_(event.snapshot);
} else {
const result = jsonpatch.applyPatch(state, structuredClone_(event.delta), true, false);
state = result.newDocument;
}
}
compacted.push({ type: EventType.STATE_SNAPSHOT, snapshot: state });
return state;
}
```

In `compactEvents`, track the running state:

```ts
let runningState: any = {};
// ...in each flushState call site:
runningState = flushState(stateEvents, compacted, runningState);
stateEvents = [];
```

STATE_SNAPSHOT still replaces wholesale (existing behaviour); STATE_DELTA applies on top of whatever the prior batch produced. The pathological case (delta against missing path) becomes a real protocol error to surface, not a consequence of an internal optimisation.

## Suggested tests

1. Round-trip the multi-run repro above through `compactEvents` — must NOT throw.
2. Two consecutive runs where run 2 has only STATE_DELTAs that target paths from run 1's snapshot — assert the compacted run-2 snapshot includes run-1 paths plus run-2 mutations.
3. Regression: single-snapshot single-batch flow produces identical output to today.

## Why this matters beyond our workaround

The current behaviour silently penalises any backend that emits high-volume state events: the only safe wire shape becomes "every delta carries the full reactive field," which is `O(N^2)` per turn. For frameworks like CopilotKit (which uses `@ag-ui/client` as transport for per-run state to React components), that means up to ~1 MB / turn for a 100-event run. Fixing `flushState` to seed lets backends use cheap `add /run_log/-` style ops again.

Per CONTRIBUTING.md I'm filing this issue first and would be happy to send the PR once a code owner can take a look (and confirm the running-state semantics match your intent, or suggest an alternative — e.g. surfacing the missing-path case as an explicit protocol error). Tagged: state management, consolidator, JSON Patch, performance.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.