diff --git a/go.mod b/go.mod
index 9a345af1..0e724529 100644
--- a/go.mod
+++ b/go.mod
@@ -8,7 +8,7 @@ require (
 	github.com/gregjones/httpcache v0.0.0-20171119193500-2bcd89a1743f
 	github.com/launchdarkly/ccache v1.1.0
 	github.com/launchdarkly/eventsource v1.10.0
-	github.com/launchdarkly/go-jsonstream/v3 v3.1.1
+	github.com/launchdarkly/go-jsonstream/v3 v3.1.2-0.20260729181618-28aa1e35cda3
 	github.com/launchdarkly/go-ntlm-proxy-auth v1.0.3
 	github.com/launchdarkly/go-sdk-common/v3 v3.5.0
 	github.com/launchdarkly/go-sdk-events/v3 v3.6.2
diff --git a/go.sum b/go.sum
index 3f81bb71..51e6ba63 100644
--- a/go.sum
+++ b/go.sum
@@ -19,8 +19,8 @@ github.com/launchdarkly/ccache v1.1.0 h1:voD1M+ZJXR3MREOKtBwgTF9hYHl1jg+vFKS/+VA
 github.com/launchdarkly/ccache v1.1.0/go.mod h1:TlxzrlnzvYeXiLHmesMuvoZetu4Z97cV1SsdqqBJi1Q=
 github.com/launchdarkly/eventsource v1.10.0 h1:H9Tp6AfGu/G2qzBJC26iperrvwhzdbiA/gx7qE2nDFI=
 github.com/launchdarkly/eventsource v1.10.0/go.mod h1:J3oa50bPvJesZqNAJtb5btSIo5N6roDWhiAS3IpsKck=
-github.com/launchdarkly/go-jsonstream/v3 v3.1.1 h1:ugupp2eNtwVbr69KCdeUrm1vUf1/3ju4Wdliaob95uY=
-github.com/launchdarkly/go-jsonstream/v3 v3.1.1/go.mod h1:ZBjhKq8mhArCtqotGRGnteY6eXpNm1GaOdUSZHh+ZjM=
+github.com/launchdarkly/go-jsonstream/v3 v3.1.2-0.20260729181618-28aa1e35cda3 h1:wT4WhK2sqM4j0nJwa+eO7g+XCblsT5myDhiadFaXeA8=
+github.com/launchdarkly/go-jsonstream/v3 v3.1.2-0.20260729181618-28aa1e35cda3/go.mod h1:ZBjhKq8mhArCtqotGRGnteY6eXpNm1GaOdUSZHh+ZjM=
 github.com/launchdarkly/go-ntlm-proxy-auth v1.0.3 h1:i3V0N+R0Fd2nXfGEVKCBIZ8kyttZ+SRKvBG8cdcphO4=
 github.com/launchdarkly/go-ntlm-proxy-auth v1.0.3/go.mod h1:kU5uMfNSTpYE6fIzmAXjFxUdmnaDPUEQ5zKm3RVKUsY=
 github.com/launchdarkly/go-ntlmssp v1.0.3 h1:rFxOnnEJ2DzJ+NU0plhXqnldJUwn3wWJFTWKCmaiQdE=
diff --git a/internal/datasourcev2/event_parsing.go b/internal/datasourcev2/event_parsing.go
new file mode 100644
index 00000000..7428ee6e
--- /dev/null
+++ b/internal/datasourcev2/event_parsing.go
@@ -0,0 +1,51 @@
+package datasourcev2
+
+import (
+	"encoding/json"
+
+	"github.com/launchdarkly/go-server-sdk/v7/subsystems"
+	"github.com/launchdarkly/go-server-sdk/v7/subsystems/ldstoretypes"
+)
+
+// This file contains the parts of FDv2 protocol event decoding that are shared by both build
+// variants. The event decoders themselves are split by build tag:
+//
+//   - event_parsing_default.go: a single-pass jsonstream decoder using jreader.RawValue/Offset for
+//     zero-copy raw capture. Each event, and in particular each put-object item's JSON, is scanned
+//     once. Its scalar decoders are written to match encoding/json's acceptance rules (see below).
+//   - event_parsing_easyjson.go: a reflection-based decode (encoding/json) for the
+//     launchdarkly_easyjson build, so that no new code depends on the easyjson token reader, which
+//     is planned for removal from go-jsonstream.
+//
+// The two variants accept and reject the same payloads. The easyjson build decodes each scalar
+// event with encoding/json into the subsystems types; the default build's jsonstream decoders are
+// written to mirror that behavior exactly -- including rejecting non-integer version/target values
+// on the plain-int types and matching Selector's float64-based (truncating) version handling. Both
+// produce a parsedPutObject carrying the item's raw bytes and, for recognized kinds, its parsed
+// form, so the ChangeSet's collections are assembled without a later re-parse.
+
+const errNoKnownPollingEvents = "didn't receive any known protocol events in polling payload"
+
+// parsedPutObject is the result of decoding a put-object event's data.
+type parsedPutObject struct {
+	kind    subsystems.ObjectKind
+	key     string
+	version int
+	// object is the item's raw JSON. In the default build it references the input buffer passed
+	// to parsePutObjectEventData and is valid only as long as that buffer is.
+	object json.RawMessage
+	// item is the parsed representation of object, set only when hasItem is true. hasItem is
+	// false when the kind is unrecognized, in which case the raw bytes are still retained for
+	// forwards compatibility.
+	item    ldstoretypes.ItemDescriptor
+	hasItem bool
+}
+
+// addTo adds the put to a change-set builder, carrying the parsed item along when there is one.
+func (p parsedPutObject) addTo(builder *subsystems.ChangeSetBuilder) {
+	if p.hasItem {
+		builder.AddParsedPut(p.kind, p.key, p.version, p.object, p.item)
+	} else {
+		builder.AddPut(p.kind, p.key, p.version, p.object)
+	}
+}
diff --git a/internal/datasourcev2/event_parsing_benchmark_test.go b/internal/datasourcev2/event_parsing_benchmark_test.go
new file mode 100644
index 00000000..cf4ae1a4
--- /dev/null
+++ b/internal/datasourcev2/event_parsing_benchmark_test.go
@@ -0,0 +1,43 @@
+package datasourcev2
+
+import (
+	"context"
+	"testing"
+)
+
+// These benchmarks guard the single-pass polling-payload parser against regressions, and keep the
+// previous reflection-based decode path measurable for comparison. The dominant cost of client
+// initialization over FDv2 is parsing the initial polling payload, so this path is
+// performance-sensitive: every SDK start pays it once per payload.
+
+func BenchmarkParsePollingPayload(b *testing.B) {
+	body := makePollingBody(b, 2000)
+	b.SetBytes(int64(len(body)))
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		changeSet, err := parsePollingPayload(context.Background(), body)
+		if err != nil {
+			b.Fatal(err)
+		}
+		if _, err := changeSet.Collections(); err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+func BenchmarkParsePollingPayloadReflectionReference(b *testing.B) {
+	body := makePollingBody(b, 2000)
+	b.SetBytes(int64(len(body)))
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		changeSet, err := parsePollingPayloadReflectionReference(body)
+		if err != nil {
+			b.Fatal(err)
+		}
+		if _, err := changeSet.Collections(); err != nil {
+			b.Fatal(err)
+		}
+	}
+}
diff --git a/internal/datasourcev2/event_parsing_default.go b/internal/datasourcev2/event_parsing_default.go
new file mode 100644
index 00000000..b6c505ff
--- /dev/null
+++ b/internal/datasourcev2/event_parsing_default.go
@@ -0,0 +1,437 @@
+//go:build !launchdarkly_easyjson
+// +build !launchdarkly_easyjson
+
+package datasourcev2
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strconv"
+
+	"github.com/launchdarkly/go-jsonstream/v3/jreader"
+	"github.com/launchdarkly/go-server-sdk/v7/subsystems"
+)
+
+// JSON property names used by the single-pass event decoders.
+const (
+	propEvents     = "events"
+	propEvent      = "event"
+	propData       = "data"
+	propKind       = "kind"
+	propKey        = "key"
+	propVersion    = "version"
+	propObject     = "object"
+	propReason     = "reason"
+	propState      = "state"
+	propPayloads   = "payloads"
+	propID         = "id"
+	propTarget     = "target"
+	propIntentCode = "intentCode"
+	propPayloadID  = "payloadId"
+)
+
+// readInt reads a JSON number that must fit in a Go int, with the same acceptance rules as decoding
+// into an int via encoding/json (the reflection decode used by the launchdarkly_easyjson build for
+// the plain-int fields), so the two build variants stay in agreement. jreader.Int cannot be used:
+// it coerces via int(Float64()), which silently truncates a fractional number (1.9 -> 1), loses
+// precision beyond 2^53, and is implementation-defined out of range. Parsing the raw number literal
+// instead means a fractional value (1.9), exponent notation (1e2), or an out-of-range value is
+// rejected, large integers up to the int range are preserved exactly, and a JSON null yields the
+// zero value -- exactly as encoding/json does.
+func readInt(r *jreader.Reader) int {
+	raw := r.RawValue()
+	if r.Error() != nil {
+		return 0
+	}
+	if string(raw) == "null" {
+		// encoding/json leaves the zero value for a null; match that rather than erroring.
+		return 0
+	}
+	n, err := strconv.ParseInt(string(raw), 10, 0)
+	if err != nil {
+		r.AddError(jreader.SyntaxError{Message: "expected integer"})
+		return 0
+	}
+	return int(n)
+}
+
+// skipValue discards the next JSON value. It deliberately uses RawValue rather than
+// jreader.SkipValue: SkipValue recurses one stack frame per level of nesting with no depth limit,
+// so a deeply nested value in an unknown field of untrusted input could overflow the stack and
+// crash the process. RawValue scans the value iteratively and validates it with encoding/json,
+// whose nesting is capped -- turning an over-deep value into an error rather than a crash, matching
+// the encoding/json decode the SDK used before this parser.
+func skipValue(r *jreader.Reader) {
+	_ = r.RawValue()
+}
+
+// This file contains the single-pass payload decoders used by the default build. Each event --
+// and in particular each put-object's item JSON -- is scanned only once: scalars are read in
+// place, and a recognized item's model is decoded directly from the stream while its raw bytes
+// are captured as a zero-copy slice of the input by recording the reader's offset around the
+// decode (the RFC 8259 compliant tokenizer fully validates everything it parses, so the captured
+// span is known-valid JSON). Only values that are never model-parsed here -- objects of
+// unrecognized kinds, and event data that arrives before its event name -- go through
+// jreader.RawValue, which performs its own boundary scan and validation.
+
+// parsePollingPayload walks a polling response body of the form {"events":[...]} in a single
+// jsonstream pass and returns the completed change set. It mirrors the event semantics of the
+// streaming data source: a server-intent of "none" short-circuits to a no-changes result, and a
+// payload-transferred event completes the change set. Unknown event names are ignored for
+// forwards compatibility.
+//
+// The returned ChangeSet's raw change objects reference the body slice, so the body must not be
+// reused or modified while the ChangeSet is alive.
+func parsePollingPayload(ctx context.Context, body []byte) (*subsystems.ChangeSet, error) {
+	builder := subsystems.NewChangeSetBuilder()
+	r := jreader.NewReader(body)
+	for topObj := r.Object(); topObj.Next(); {
+		if string(topObj.Name()) != propEvents {
+			skipValue(&r)
+			continue
+		}
+		for arr := r.Array(); arr.Next(); {
+			select {
+			case <-ctx.Done():
+				return nil, ctx.Err()
+			default:
+			}
+			changeSet, done, err := readPollingEvent(&r, builder)
+			if err != nil {
+				return nil, err
+			}
+			if done {
+				return changeSet, nil
+			}
+		}
+	}
+	if err := r.Error(); err != nil {
+		return nil, err
+	}
+	return nil, errors.New(errNoKnownPollingEvents)
+}
+
+// readPollingEvent consumes one {"event":...,"data":...} object from the events array. When the
+// event completes the payload (server-intent "none" or payload-transferred), it returns the
+// resulting change set with done set to true.
+//
+// The event's data is captured verbatim and decoded once the whole event object has been read.
+// This keeps the decode independent of property order (the event name may follow the data) and
+// makes a duplicated data property resolve last-wins, exactly as the reflection-based decode does.
+func readPollingEvent(
+	r *jreader.Reader,
+	builder *subsystems.ChangeSetBuilder,
+) (changeSet *subsystems.ChangeSet, done bool, err error) {
+	var name subsystems.EventName
+	var data []byte
+	gotData := false
+	for obj := r.Object(); obj.Next(); {
+		switch string(obj.Name()) {
+		case propEvent:
+			name = subsystems.EventName(r.String())
+		case propData:
+			gotData = true
+			data = r.RawValue()
+		default:
+			skipValue(r)
+		}
+	}
+	if err := r.Error(); err != nil {
+		return nil, false, err
+	}
+	if !gotData {
+		switch name {
+		case subsystems.EventServerIntent, subsystems.EventPutObject,
+			subsystems.EventDeleteObject, subsystems.EventPayloadTransferred:
+			// A payload-affecting event without data cannot be applied; treat the payload as
+			// malformed rather than silently dropping the event.
+			return nil, false, fmt.Errorf("polling payload event %q has no data", name)
+		}
+		return nil, false, nil
+	}
+
+	dataReader := jreader.NewReader(data)
+	switch name {
+	case subsystems.EventServerIntent:
+		intent, err := readServerIntent(&dataReader)
+		if err != nil {
+			return nil, false, err
+		}
+		if intent.Payload.Code == subsystems.IntentNone {
+			return builder.NoChanges(), true, nil
+		}
+		builder.Start(intent)
+	case subsystems.EventPutObject:
+		put, err := readPutObject(&dataReader, data)
+		if err != nil {
+			return nil, false, err
+		}
+		put.addTo(builder)
+	case subsystems.EventDeleteObject:
+		deleteObject, err := readDeleteObject(&dataReader)
+		if err != nil {
+			return nil, false, err
+		}
+		builder.AddDelete(deleteObject.Kind, deleteObject.Key, deleteObject.Version)
+	case subsystems.EventPayloadTransferred:
+		selector, err := readSelector(&dataReader)
+		if err != nil {
+			return nil, false, err
+		}
+		finished, err := builder.Finish(selector)
+		if err != nil {
+			return nil, false, err
+		}
+		return finished, true, nil
+	default:
+		// An unknown event name is ignored for forwards compatibility. RawValue already validated
+		// the data above.
+	}
+	return nil, false, nil
+}
+
+// parsePutObjectEventData decodes the data of a put-object event in a single pass, capturing the
+// item's raw JSON and its parsed form together.
+func parsePutObjectEventData(data []byte) (parsedPutObject, error) {
+	r := jreader.NewReader(data)
+	p, err := readPutObject(&r, data)
+	if err == nil {
+		err = r.Error()
+	}
+	return p, err
+}
+
+// jsonWhitespace is the set of insignificant whitespace bytes RFC 8259 allows between tokens; a
+// span captured via reader offsets may include them around the value and they are trimmed off.
+const jsonWhitespace = " \t\r\n"
+
+// readPutObject decodes a put-object event's data from r, which must have been created over
+// input (offset-based span capture slices it).
+func readPutObject(r *jreader.Reader, input []byte) (parsedPutObject, error) {
+	var p parsedPutObject
+	var object json.RawMessage
+	// decodedKind records the kind under which the item was eagerly model-decoded while reading
+	// the object, so that a "kind" property appearing after "object" (non-conformant, but json's
+	// last-wins behavior handles it) can be detected and the item re-parsed under the final kind
+	// rather than left mis-typed.
+	var decodedKind subsystems.ObjectKind
+	for obj := r.Object(); obj.Next(); {
+		switch string(obj.Name()) {
+		case propKind:
+			p.kind = subsystems.ObjectKind(r.String())
+		case propKey:
+			p.key = r.String()
+		case propVersion:
+			p.version = readInt(r)
+		case propObject:
+			if kind, recognized := p.kind.ToFDV1(); recognized {
+				// The kind is already known (LaunchDarkly services always write "kind" before
+				// "object"), so the item is model-decoded directly from the stream -- which
+				// fully validates it -- while the reader offsets around the decode capture the
+				// value's raw bytes with no additional scan.
+				start := r.Offset()
+				item, err := kind.DeserializeFromJSONReader(r)
+				if err != nil {
+					return p, err
+				}
+				span := bytes.Trim(input[start:r.Offset()], jsonWhitespace)
+				object = json.RawMessage(span[:len(span):len(span)])
+				p.item, p.hasItem, decodedKind = item, true, p.kind
+			} else {
+				// The kind is unrecognized or has not been read yet, so the value cannot be
+				// model-decoded here; RawValue captures it with full validation, and if the
+				// kind turns out to be recognized it is deserialized after the loop.
+				object = json.RawMessage(r.RawValue())
+			}
+		default:
+			skipValue(r)
+		}
+	}
+	if err := r.Error(); err != nil {
+		return p, err
+	}
+	p.object = object
+	if object == nil {
+		return p, nil
+	}
+	kind, recognized := p.kind.ToFDV1()
+	if !recognized {
+		// The final kind is unrecognized -- either it was never a known kind, or a later "kind"
+		// property overrode an earlier recognized one. Keep the object raw and unparsed for
+		// forwards compatibility, discarding any item eagerly decoded under the earlier kind. The
+		// bytes were already validated (by the eager decode or by RawValue).
+		p.hasItem = false
+		return p, nil
+	}
+	if p.hasItem && decodedKind == p.kind {
+		// The item was already decoded under the final kind while reading the object.
+		return p, nil
+	}
+	// The object was captured before its kind was known, or a later "kind" changed the type after
+	// the object was decoded; parse the captured bytes under the final kind so that the stored item
+	// and kind always agree.
+	itemReader := jreader.NewReader(object)
+	item, err := kind.DeserializeFromJSONReader(&itemReader)
+	if err != nil {
+		return p, err
+	}
+	p.item, p.hasItem = item, true
+	return p, nil
+}
+
+// parseServerIntentEventData decodes the data of a server-intent event. The intent is required to
+// have at least one payload (at index 0) at this time, matching subsystems.ServerIntent's decode.
+func parseServerIntentEventData(data []byte) (subsystems.ServerIntent, error) {
+	r := jreader.NewReader(data)
+	intent, err := readServerIntent(&r)
+	if err == nil {
+		err = r.Error()
+	}
+	return intent, err
+}
+
+func readServerIntent(r *jreader.Reader) (subsystems.ServerIntent, error) {
+	var intent subsystems.ServerIntent
+	gotPayload := false
+	for obj := r.Object(); obj.Next(); {
+		if string(obj.Name()) != propPayloads {
+			skipValue(r)
+			continue
+		}
+		for arr := r.Array(); arr.Next(); {
+			// The protocol allows more than one payload, but SDKs currently only support one;
+			// any additional payloads are skipped.
+			if gotPayload {
+				skipValue(r)
+				continue
+			}
+			for payloadObj := r.Object(); payloadObj.Next(); {
+				switch string(payloadObj.Name()) {
+				case propID:
+					intent.Payload.ID = r.String()
+				case propTarget:
+					intent.Payload.Target = readInt(r)
+				case propIntentCode:
+					intent.Payload.Code = subsystems.IntentCode(r.String())
+				case propReason:
+					intent.Payload.Reason = r.String()
+				default:
+					skipValue(r)
+				}
+			}
+			gotPayload = true
+		}
+	}
+	if err := r.Error(); err != nil {
+		return intent, err
+	}
+	if !gotPayload {
+		// It is a protocol error for the payload list to be missing or empty.
+		return intent, errors.New("changeset: server-intent event has no payloads")
+	}
+	return intent, nil
+}
+
+// parseDeleteObjectEventData decodes the data of a delete-object event.
+func parseDeleteObjectEventData(data []byte) (subsystems.DeleteObject, error) {
+	r := jreader.NewReader(data)
+	deleteObject, err := readDeleteObject(&r)
+	if err == nil {
+		err = r.Error()
+	}
+	return deleteObject, err
+}
+
+func readDeleteObject(r *jreader.Reader) (subsystems.DeleteObject, error) {
+	var d subsystems.DeleteObject
+	for obj := r.Object(); obj.Next(); {
+		switch string(obj.Name()) {
+		case propKind:
+			d.Kind = subsystems.ObjectKind(r.String())
+		case propKey:
+			d.Key = r.String()
+		case propVersion:
+			d.Version = readInt(r)
+		default:
+			skipValue(r)
+		}
+	}
+	return d, r.Error()
+}
+
+// parseSelectorEventData decodes the data of a payload-transferred event. Both the state and the
+// version are required.
+func parseSelectorEventData(data []byte) (subsystems.Selector, error) {
+	r := jreader.NewReader(data)
+	selector, err := readSelector(&r)
+	if err == nil {
+		err = r.Error()
+	}
+	return selector, err
+}
+
+func readSelector(r *jreader.Reader) (subsystems.Selector, error) {
+	var state string
+	var version int
+	gotState, gotVersion := false, false
+	for obj := r.Object(); obj.Next(); {
+		switch string(obj.Name()) {
+		case propState:
+			state = r.String()
+			gotState = true
+		case propVersion:
+			// subsystems.Selector's decode (the reflection reference for this event) reads version
+			// through float64 and truncates, so use the lenient jreader.Int here rather than the
+			// strict readInt used for the other integer fields, keeping the two builds in
+			// agreement. The value is deprecated and not used by consumers.
+			version = r.Int()
+			gotVersion = true
+		default:
+			skipValue(r)
+		}
+	}
+	if err := r.Error(); err != nil {
+		return subsystems.NoSelector(), err
+	}
+	if !gotState {
+		return subsystems.NoSelector(), errors.New("unmarshal selector: missing state field")
+	}
+	if !gotVersion {
+		return subsystems.NoSelector(), errors.New("unmarshal selector: missing version field")
+	}
+	return subsystems.NewSelector(state, version), nil
+}
+
+// parseGoodbyeEventData decodes the data of a goodbye event.
+func parseGoodbyeEventData(data []byte) (subsystems.Goodbye, error) {
+	r := jreader.NewReader(data)
+	var goodbye subsystems.Goodbye
+	for obj := r.Object(); obj.Next(); {
+		if string(obj.Name()) == propReason {
+			goodbye.Reason = r.String()
+		} else {
+			skipValue(&r)
+		}
+	}
+	return goodbye, r.Error()
+}
+
+// parseErrorEventData decodes the data of an error event.
+func parseErrorEventData(data []byte) (subsystems.Error, error) {
+	r := jreader.NewReader(data)
+	var errorData subsystems.Error
+	for obj := r.Object(); obj.Next(); {
+		switch string(obj.Name()) {
+		case propPayloadID:
+			errorData.PayloadID = r.String()
+		case propReason:
+			errorData.Reason = r.String()
+		default:
+			skipValue(&r)
+		}
+	}
+	return errorData, r.Error()
+}
diff --git a/internal/datasourcev2/event_parsing_default_test.go b/internal/datasourcev2/event_parsing_default_test.go
new file mode 100644
index 00000000..458bd7a5
--- /dev/null
+++ b/internal/datasourcev2/event_parsing_default_test.go
@@ -0,0 +1,28 @@
+//go:build !launchdarkly_easyjson
+// +build !launchdarkly_easyjson
+
+package datasourcev2
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// These tests document behavior specific to the default build's single-pass parser. The easyjson
+// build's reflection-based decode parses the whole envelope up front, so it does not share them.
+
+func TestParsePollingPayloadStopsReadingAfterTransfer(t *testing.T) {
+	// Once payload-transferred completes the change set, the single-pass parser never reads the
+	// rest of the body -- even content that is not valid JSON.
+	body := []byte(`{"events":[` +
+		`{"event":"server-intent","data":{"payloads":[{"id":"p1","target":1,"intentCode":"xfer-full","reason":"r"}]}},` +
+		`{"event":"payload-transferred","data":{"state":"s","version":1}},` +
+		`{"event":"put-object","data":{"this is": "not even valid put data`,
+	)
+	changeSet, err := parsePollingPayload(context.Background(), body)
+	require.NoError(t, err)
+	assert.Empty(t, changeSet.Changes())
+}
diff --git a/internal/datasourcev2/event_parsing_easyjson.go b/internal/datasourcev2/event_parsing_easyjson.go
new file mode 100644
index 00000000..e36643d3
--- /dev/null
+++ b/internal/datasourcev2/event_parsing_easyjson.go
@@ -0,0 +1,134 @@
+//go:build launchdarkly_easyjson
+// +build launchdarkly_easyjson
+
+package datasourcev2
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+
+	"github.com/launchdarkly/go-jsonstream/v3/jreader"
+	"github.com/launchdarkly/go-server-sdk/v7/subsystems"
+)
+
+// This file contains the payload decoders used by the launchdarkly_easyjson build. It retains
+// the reflection-based decode (encoding/json over the polling envelope and each put-object
+// event) so that nothing in this build depends on jreader.RawValue: easyjson support is planned
+// for removal from go-jsonstream, and no new code should rely on its token reader.
+//
+// The results are identical to the default build's single-pass decoder -- including the eager
+// item deserialization that lets the ChangeSet pre-populate its collections -- at the cost of
+// the extra reflection scans and byte copies that the default build avoids.
+
+// parsePollingPayload decodes a polling response body of the form {"events":[...]} and returns
+// the completed change set. Refer to the default build's implementation for the event semantics;
+// the two are behaviorally equivalent.
+func parsePollingPayload(ctx context.Context, body []byte) (*subsystems.ChangeSet, error) {
+	var payload subsystems.PollingPayload
+	if err := json.Unmarshal(body, &payload); err != nil {
+		return nil, err
+	}
+	builder := subsystems.NewChangeSetBuilder()
+	for _, event := range payload.Events {
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		default:
+		}
+		switch event.Name {
+		case subsystems.EventServerIntent:
+			intent, err := parseServerIntentEventData(event.Data)
+			if err != nil {
+				return nil, err
+			}
+			if intent.Payload.Code == subsystems.IntentNone {
+				return builder.NoChanges(), nil
+			}
+			builder.Start(intent)
+		case subsystems.EventPutObject:
+			put, err := parsePutObjectEventData(event.Data)
+			if err != nil {
+				return nil, err
+			}
+			put.addTo(builder)
+		case subsystems.EventDeleteObject:
+			deleteObject, err := parseDeleteObjectEventData(event.Data)
+			if err != nil {
+				return nil, err
+			}
+			builder.AddDelete(deleteObject.Kind, deleteObject.Key, deleteObject.Version)
+		case subsystems.EventPayloadTransferred:
+			selector, err := parseSelectorEventData(event.Data)
+			if err != nil {
+				return nil, err
+			}
+			return builder.Finish(selector)
+		default:
+			// An unknown event name is ignored for forwards compatibility.
+		}
+	}
+	return nil, errors.New(errNoKnownPollingEvents)
+}
+
+// parsePutObjectEventData decodes the data of a put-object event via encoding/json, then eagerly
+// deserializes the item so the change set can carry the parsed form alongside the raw bytes.
+func parsePutObjectEventData(data []byte) (parsedPutObject, error) {
+	var put subsystems.PutObject
+	if err := json.Unmarshal(data, &put); err != nil {
+		return parsedPutObject{}, err
+	}
+	p := parsedPutObject{kind: put.Kind, key: put.Key, version: put.Version, object: put.Object}
+	kind, recognized := put.Kind.ToFDV1()
+	if !recognized || put.Object == nil {
+		// An unrecognized kind is kept raw for forwards compatibility; its JSON syntax was
+		// already validated by the json.RawMessage decode above.
+		return p, nil
+	}
+	itemReader := jreader.NewReader(put.Object)
+	item, err := kind.DeserializeFromJSONReader(&itemReader)
+	if err != nil {
+		return p, err
+	}
+	p.item, p.hasItem = item, true
+	return p, nil
+}
+
+// The scalar event decoders below use encoding/json into the subsystems types. This is the
+// reflection decode the default build's jsonstream decoders are written to match; keeping it here
+// means the launchdarkly_easyjson build depends only on encoding/json for these events.
+
+// parseServerIntentEventData decodes the data of a server-intent event.
+func parseServerIntentEventData(data []byte) (subsystems.ServerIntent, error) {
+	var intent subsystems.ServerIntent
+	err := json.Unmarshal(data, &intent)
+	return intent, err
+}
+
+// parseDeleteObjectEventData decodes the data of a delete-object event.
+func parseDeleteObjectEventData(data []byte) (subsystems.DeleteObject, error) {
+	var deleteObject subsystems.DeleteObject
+	err := json.Unmarshal(data, &deleteObject)
+	return deleteObject, err
+}
+
+// parseSelectorEventData decodes the data of a payload-transferred event.
+func parseSelectorEventData(data []byte) (subsystems.Selector, error) {
+	var selector subsystems.Selector
+	err := json.Unmarshal(data, &selector)
+	return selector, err
+}
+
+// parseGoodbyeEventData decodes the data of a goodbye event.
+func parseGoodbyeEventData(data []byte) (subsystems.Goodbye, error) {
+	var goodbye subsystems.Goodbye
+	err := json.Unmarshal(data, &goodbye)
+	return goodbye, err
+}
+
+// parseErrorEventData decodes the data of an error event.
+func parseErrorEventData(data []byte) (subsystems.Error, error) {
+	var errorData subsystems.Error
+	err := json.Unmarshal(data, &errorData)
+	return errorData, err
+}
diff --git a/internal/datasourcev2/event_parsing_test.go b/internal/datasourcev2/event_parsing_test.go
new file mode 100644
index 00000000..01e3f212
--- /dev/null
+++ b/internal/datasourcev2/event_parsing_test.go
@@ -0,0 +1,582 @@
+package datasourcev2
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/launchdarkly/go-sdk-common/v3/ldvalue"
+	"github.com/launchdarkly/go-server-sdk-evaluation/v3/ldbuilders"
+	"github.com/launchdarkly/go-server-sdk-evaluation/v3/ldmodel"
+	"github.com/launchdarkly/go-server-sdk/v7/internal/datakinds"
+	"github.com/launchdarkly/go-server-sdk/v7/subsystems"
+	"github.com/launchdarkly/go-server-sdk/v7/subsystems/ldstoretypes"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func makeTestFlagJSON(t testing.TB, key string, version int) []byte {
+	t.Helper()
+	flag := ldbuilders.NewFlagBuilder(key).Version(version).
+		On(true).
+		Variations(ldvalue.Bool(true), ldvalue.Bool(false)).
+		FallthroughVariation(0).
+		AddTarget(0, "user-a", "user-b").
+		AddRule(ldbuilders.NewRuleBuilder().ID("rule-1").Variation(1).Clauses(
+			ldbuilders.Clause("email", "in", ldvalue.String("test@example.com")),
+		)).
+		Build()
+	return datakinds.Features.Serialize(ldstoretypes.ItemDescriptor{Version: version, Item: &flag})
+}
+
+func makeTestSegmentJSON(t testing.TB, key string, version int) []byte {
+	t.Helper()
+	segment := ldbuilders.NewSegmentBuilder(key).Version(version).Included("user-a").Build()
+	return datakinds.Segments.Serialize(ldstoretypes.ItemDescriptor{Version: version, Item: &segment})
+}
+
+func TestParsePutObjectEventDataPropertyOrderings(t *testing.T) {
+	flagJSON := makeTestFlagJSON(t, "flagkey", 3)
+
+	kindFirst := fmt.Sprintf(`{"kind":"flag","key":"flagkey","version":3,"object":%s}`, flagJSON)
+	objectFirst := fmt.Sprintf(`{"object":%s,"version":3,"key":"flagkey","kind":"flag"}`, flagJSON)
+
+	for name, data := range map[string]string{"kind before object": kindFirst, "object before kind": objectFirst} {
+		t.Run(name, func(t *testing.T) {
+			p, err := parsePutObjectEventData([]byte(data))
+			require.NoError(t, err)
+			assert.Equal(t, subsystems.FlagKind, p.kind)
+			assert.Equal(t, "flagkey", p.key)
+			assert.Equal(t, 3, p.version)
+			assert.JSONEq(t, string(flagJSON), string(p.object))
+			require.True(t, p.hasItem)
+			assert.Equal(t, 3, p.item.Version)
+			require.IsType(t, &ldmodel.FeatureFlag{}, p.item.Item)
+			assert.Equal(t, "flagkey", p.item.Item.(*ldmodel.FeatureFlag).Key)
+		})
+	}
+}
+
+func TestParsePutObjectEventDataSegmentKind(t *testing.T) {
+	segmentJSON := makeTestSegmentJSON(t, "segmentkey", 2)
+	data := fmt.Sprintf(`{"kind":"segment","key":"segmentkey","version":2,"object":%s}`, segmentJSON)
+	p, err := parsePutObjectEventData([]byte(data))
+	require.NoError(t, err)
+	assert.Equal(t, subsystems.SegmentKind, p.kind)
+	require.True(t, p.hasItem)
+	require.IsType(t, &ldmodel.Segment{}, p.item.Item)
+	assert.Equal(t, "segmentkey", p.item.Item.(*ldmodel.Segment).Key)
+}
+
+func TestParsePutObjectEventDataUnknownKind(t *testing.T) {
+	t.Run("valid object is retained raw", func(t *testing.T) {
+		data := `{"kind":"future-kind","key":"k","version":1,"object":{"key":"k","futureProp":[1,2]}}`
+		p, err := parsePutObjectEventData([]byte(data))
+		require.NoError(t, err)
+		assert.Equal(t, subsystems.ObjectKind("future-kind"), p.kind)
+		assert.False(t, p.hasItem)
+		assert.JSONEq(t, `{"key":"k","futureProp":[1,2]}`, string(p.object))
+	})
+
+	t.Run("malformed object is still an error", func(t *testing.T) {
+		// The braces balance, so this proves the object bytes are fully validated even though
+		// an unrecognized kind is never model-parsed (in the default build that validation
+		// comes from jreader.RawValue itself; in the easyjson build, from the json.RawMessage
+		// decode).
+		data := `{"kind":"future-kind","key":"k","version":1,"object":{"bad":}}`
+		_, err := parsePutObjectEventData([]byte(data))
+		assert.Error(t, err)
+	})
+}
+
+func TestParsePutObjectEventDataMalformedObjectOfKnownKind(t *testing.T) {
+	data := `{"kind":"flag","key":"k","version":1,"object":{"key":12345,"on":"not-a-bool"}}`
+	_, err := parsePutObjectEventData([]byte(data))
+	assert.Error(t, err)
+}
+
+func TestParsePutObjectEventDataPreservesObjectBytesExactly(t *testing.T) {
+	// Whitespace surrounding the object value must not leak into the captured raw bytes, and
+	// whitespace inside the value must be preserved verbatim -- relay embeds these bytes
+	// unmodified into downstream events.
+	data := `{ "kind": "flag", "key": "flagkey", "version": 1,` + "\n\t" +
+		`"object": { "key": "flagkey", "version": 1, "on": true } }`
+	p, err := parsePutObjectEventData([]byte(data))
+	require.NoError(t, err)
+	assert.Equal(t, `{ "key": "flagkey", "version": 1, "on": true }`, string(p.object))
+	require.True(t, p.hasItem)
+	assert.Equal(t, "flagkey", p.item.Item.(*ldmodel.FeatureFlag).Key)
+}
+
+func TestParsePutObjectEventDataIgnoresExtraProperties(t *testing.T) {
+	flagJSON := makeTestFlagJSON(t, "flagkey", 1)
+	data := fmt.Sprintf(`{"kind":"flag","futureField":{"a":[1]},"key":"flagkey","version":1,"object":%s}`, flagJSON)
+	p, err := parsePutObjectEventData([]byte(data))
+	require.NoError(t, err)
+	assert.True(t, p.hasItem)
+}
+
+func TestParseServerIntentEventData(t *testing.T) {
+	t.Run("single payload", func(t *testing.T) {
+		data := `{"payloads":[{"id":"p1","target":3,"intentCode":"xfer-full","reason":"payload-missing"}]}`
+		intent, err := parseServerIntentEventData([]byte(data))
+		require.NoError(t, err)
+		assert.Equal(t, "p1", intent.Payload.ID)
+		assert.Equal(t, 3, intent.Payload.Target)
+		assert.Equal(t, subsystems.IntentTransferFull, intent.Payload.Code)
+		assert.Equal(t, "payload-missing", intent.Payload.Reason)
+	})
+
+	t.Run("only the first of multiple payloads is used", func(t *testing.T) {
+		data := `{"payloads":[{"id":"p1","intentCode":"none"},{"id":"p2","intentCode":"xfer-full"}]}`
+		intent, err := parseServerIntentEventData([]byte(data))
+		require.NoError(t, err)
+		assert.Equal(t, "p1", intent.Payload.ID)
+		assert.Equal(t, subsystems.IntentNone, intent.Payload.Code)
+	})
+
+	t.Run("empty payloads is an error", func(t *testing.T) {
+		_, err := parseServerIntentEventData([]byte(`{"payloads":[]}`))
+		assert.Error(t, err)
+	})
+
+	t.Run("missing payloads is an error", func(t *testing.T) {
+		_, err := parseServerIntentEventData([]byte(`{}`))
+		assert.Error(t, err)
+	})
+}
+
+func TestParseDeleteObjectEventData(t *testing.T) {
+	data := `{"version":7,"kind":"segment","key":"gone"}`
+	d, err := parseDeleteObjectEventData([]byte(data))
+	require.NoError(t, err)
+	assert.Equal(t, subsystems.DeleteObject{Version: 7, Kind: subsystems.SegmentKind, Key: "gone"}, d)
+}
+
+func TestParseSelectorEventData(t *testing.T) {
+	t.Run("valid", func(t *testing.T) {
+		selector, err := parseSelectorEventData([]byte(`{"state":"s1","version":9}`))
+		require.NoError(t, err)
+		assert.Equal(t, "s1", selector.State())
+		assert.Equal(t, 9, selector.Version())
+	})
+
+	t.Run("missing state is an error", func(t *testing.T) {
+		_, err := parseSelectorEventData([]byte(`{"version":9}`))
+		assert.Error(t, err)
+	})
+
+	t.Run("missing version is an error", func(t *testing.T) {
+		_, err := parseSelectorEventData([]byte(`{"state":"s1"}`))
+		assert.Error(t, err)
+	})
+}
+
+func TestParseGoodbyeEventData(t *testing.T) {
+	goodbye, err := parseGoodbyeEventData([]byte(`{"reason":"see ya","silent":true}`))
+	require.NoError(t, err)
+	assert.Equal(t, "see ya", goodbye.Reason)
+}
+
+func TestParseErrorEventData(t *testing.T) {
+	errorData, err := parseErrorEventData([]byte(`{"payloadId":"p1","reason":"broke"}`))
+	require.NoError(t, err)
+	assert.Equal(t, subsystems.Error{PayloadID: "p1", Reason: "broke"}, errorData)
+}
+
+func makePollingBody(t testing.TB, flagCount int) []byte {
+	t.Helper()
+	var sb strings.Builder
+	sb.WriteString(`{"events":[`)
+	sb.WriteString(`{"event":"server-intent","data":{"payloads":[{"id":"p1","target":1,"intentCode":"xfer-full","reason":"payload-missing"}]}}`)
+	for i := 0; i < flagCount; i++ {
+		fmt.Fprintf(&sb, `,{"event":"put-object","data":{"kind":"flag","key":"flag-%d","version":%d,"object":%s}}`,
+			i, i+1, makeTestFlagJSON(t, fmt.Sprintf("flag-%d", i), i+1))
+	}
+	fmt.Fprintf(&sb, `,{"event":"put-object","data":{"kind":"segment","key":"segment-0","version":1,"object":%s}}`,
+		makeTestSegmentJSON(t, "segment-0", 1))
+	sb.WriteString(`,{"event":"delete-object","data":{"kind":"flag","key":"deleted-flag","version":99}}`)
+	sb.WriteString(`,{"event":"payload-transferred","data":{"state":"p1:10","version":10}}`)
+	sb.WriteString(`]}`)
+	return []byte(sb.String())
+}
+
+func TestParsePollingPayloadFullTransfer(t *testing.T) {
+	body := makePollingBody(t, 2)
+	changeSet, err := parsePollingPayload(context.Background(), body)
+	require.NoError(t, err)
+
+	assert.Equal(t, subsystems.IntentTransferFull, changeSet.IntentCode())
+	assert.Equal(t, "p1:10", changeSet.Selector().State())
+	assert.Equal(t, 10, changeSet.Selector().Version())
+
+	changes := changeSet.Changes()
+	require.Len(t, changes, 4) // 2 flags + 1 segment + 1 delete
+	assert.Equal(t, subsystems.ChangeTypePut, changes[0].Action)
+	assert.Equal(t, "flag-0", changes[0].Key)
+	assert.JSONEq(t, string(makeTestFlagJSON(t, "flag-0", 1)), string(changes[0].Object))
+	assert.Equal(t, subsystems.ChangeTypeDelete, changes[3].Action)
+	assert.Equal(t, "deleted-flag", changes[3].Key)
+
+	collections, err := changeSet.Collections()
+	require.NoError(t, err)
+	itemsByKind := map[string]int{}
+	for _, coll := range collections {
+		itemsByKind[coll.Kind.GetName()] = len(coll.Items)
+	}
+	assert.Equal(t, map[string]int{"features": 3, "segments": 1}, itemsByKind) // 2 puts + 1 tombstone
+	for _, coll := range collections {
+		if coll.Kind.GetName() != "features" {
+			continue
+		}
+		for _, item := range coll.Items {
+			if item.Key == "deleted-flag" {
+				assert.Nil(t, item.Item.Item)
+				assert.Equal(t, 99, item.Item.Version)
+			} else {
+				assert.NotNil(t, item.Item.Item)
+			}
+		}
+	}
+}
+
+func TestParsePollingPayloadIntentNone(t *testing.T) {
+	body := []byte(`{"events":[{"event":"server-intent","data":{"payloads":[{"id":"p1","target":1,"intentCode":"none","reason":"up-to-date"}]}}]}`)
+	changeSet, err := parsePollingPayload(context.Background(), body)
+	require.NoError(t, err)
+	assert.Equal(t, subsystems.IntentNone, changeSet.IntentCode())
+	assert.Empty(t, changeSet.Changes())
+}
+
+func TestParsePollingPayloadNoKnownEventsIsError(t *testing.T) {
+	for name, body := range map[string]string{
+		"empty events":        `{"events":[]}`,
+		"no events property":  `{"other":true}`,
+		"only unknown events": `{"events":[{"event":"future-event","data":{"a":1}}]}`,
+	} {
+		t.Run(name, func(t *testing.T) {
+			_, err := parsePollingPayload(context.Background(), []byte(body))
+			assert.Error(t, err)
+		})
+	}
+}
+
+func TestParsePollingPayloadMalformedBody(t *testing.T) {
+	for _, body := range []string{
+		`{`,
+		`[]`,
+		`{"events":[{"event":"server-intent","data":{"payloads":[{}]}}`,
+		`{"events":[{"event":"put-object","data":{"kind":"flag","key":"k","version":1,"object":{"key":}}}]}`,
+	} {
+		t.Run(body, func(t *testing.T) {
+			_, err := parsePollingPayload(context.Background(), []byte(body))
+			assert.Error(t, err)
+		})
+	}
+}
+
+func TestParsePollingPayloadUnknownEventsAreIgnored(t *testing.T) {
+	body := []byte(`{"events":[` +
+		`{"event":"future-event","data":{"whatever":[1,2,3]}},` +
+		`{"event":"server-intent","data":{"payloads":[{"id":"p1","target":1,"intentCode":"xfer-full","reason":"r"}]}},` +
+		`{"event":"another-future-event","data":"scalar data"},` +
+		`{"event":"payload-transferred","data":{"state":"s","version":1}}` +
+		`]}`)
+	changeSet, err := parsePollingPayload(context.Background(), body)
+	require.NoError(t, err)
+	assert.Equal(t, subsystems.IntentTransferFull, changeSet.IntentCode())
+	assert.Equal(t, "s", changeSet.Selector().State())
+}
+
+func TestParsePollingPayloadDataBeforeEventName(t *testing.T) {
+	flagJSON := makeTestFlagJSON(t, "flagkey", 1)
+	body := []byte(fmt.Sprintf(`{"events":[`+
+		`{"data":{"payloads":[{"id":"p1","target":1,"intentCode":"xfer-full","reason":"r"}]},"event":"server-intent"},`+
+		`{"data":{"kind":"flag","key":"flagkey","version":1,"object":%s},"event":"put-object"},`+
+		`{"data":{"state":"s","version":1},"event":"payload-transferred"}`+
+		`]}`, flagJSON))
+	changeSet, err := parsePollingPayload(context.Background(), body)
+	require.NoError(t, err)
+	changes := changeSet.Changes()
+	require.Len(t, changes, 1)
+	assert.Equal(t, "flagkey", changes[0].Key)
+	assert.JSONEq(t, string(flagJSON), string(changes[0].Object))
+	collections, err := changeSet.Collections()
+	require.NoError(t, err)
+	require.Len(t, collections, 1)
+	require.Len(t, collections[0].Items, 1)
+	assert.NotNil(t, collections[0].Items[0].Item.Item)
+}
+
+func TestParsePollingPayloadEventWithoutDataIsError(t *testing.T) {
+	body := []byte(`{"events":[` +
+		`{"event":"server-intent","data":{"payloads":[{"id":"p1","target":1,"intentCode":"xfer-full","reason":"r"}]}},` +
+		`{"event":"put-object"},` +
+		`{"event":"payload-transferred","data":{"state":"s","version":1}}` +
+		`]}`)
+	_, err := parsePollingPayload(context.Background(), body)
+	assert.Error(t, err)
+}
+
+func TestParsePollingPayloadEventsAfterTransferAreIgnored(t *testing.T) {
+	// The trailing put-object is valid JSON but is never dispatched, because the payload is
+	// already complete. (The default build's single-pass parser stops reading the body entirely
+	// at that point; the easyjson build's envelope decode still requires the remainder to be
+	// well-formed JSON, which is why this event must be syntactically valid.)
+	body := []byte(`{"events":[` +
+		`{"event":"server-intent","data":{"payloads":[{"id":"p1","target":1,"intentCode":"xfer-full","reason":"r"}]}},` +
+		`{"event":"payload-transferred","data":{"state":"s","version":1}},` +
+		`{"event":"put-object","data":{"kind":"flag","key":"never-dispatched","version":1,"object":{"key":12345}}}` +
+		`]}`,
+	)
+	changeSet, err := parsePollingPayload(context.Background(), body)
+	require.NoError(t, err)
+	assert.Empty(t, changeSet.Changes())
+}
+
+func TestParsePollingPayloadHonorsContextCancellation(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	_, err := parsePollingPayload(ctx, makePollingBody(t, 2))
+	assert.ErrorIs(t, err, context.Canceled)
+}
+
+// parsePollingPayloadReflectionReference replicates the previous reflection-based parse
+// (PollingPayload envelope unmarshal, per-event unmarshal, raw puts) so that tests and
+// benchmarks can verify the single-pass parser against it.
+func parsePollingPayloadReflectionReference(body []byte) (*subsystems.ChangeSet, error) {
+	var payload subsystems.PollingPayload
+	if err := json.Unmarshal(body, &payload); err != nil {
+		return nil, err
+	}
+	builder := subsystems.NewChangeSetBuilder()
+	for _, event := range payload.Events {
+		switch event.Name {
+		case subsystems.EventServerIntent:
+			var serverIntent subsystems.ServerIntent
+			if err := json.Unmarshal(event.Data, &serverIntent); err != nil {
+				return nil, err
+			}
+			if serverIntent.Payload.Code == subsystems.IntentNone {
+				return builder.NoChanges(), nil
+			}
+			builder.Start(serverIntent)
+		case subsystems.EventPutObject:
+			var put subsystems.PutObject
+			if err := json.Unmarshal(event.Data, &put); err != nil {
+				return nil, err
+			}
+			builder.AddPut(put.Kind, put.Key, put.Version, put.Object)
+		case subsystems.EventDeleteObject:
+			var deleteObject subsystems.DeleteObject
+			if err := json.Unmarshal(event.Data, &deleteObject); err != nil {
+				return nil, err
+			}
+			builder.AddDelete(deleteObject.Kind, deleteObject.Key, deleteObject.Version)
+		case subsystems.EventPayloadTransferred:
+			var selector subsystems.Selector
+			if err := json.Unmarshal(event.Data, &selector); err != nil {
+				return nil, err
+			}
+			return builder.Finish(selector)
+		}
+	}
+	return nil, fmt.Errorf("didn't receive any known protocol events in polling payload")
+}
+
+func TestParsePollingPayloadMatchesReflectionReference(t *testing.T) {
+	body := makePollingBody(t, 20)
+
+	got, err := parsePollingPayload(context.Background(), body)
+	require.NoError(t, err)
+	want, err := parsePollingPayloadReflectionReference(body)
+	require.NoError(t, err)
+
+	assert.Equal(t, want.IntentCode(), got.IntentCode())
+	assert.Equal(t, want.Selector(), got.Selector())
+
+	wantChanges, gotChanges := want.Changes(), got.Changes()
+	require.Equal(t, len(wantChanges), len(gotChanges))
+	for i := range wantChanges {
+		assert.Equal(t, wantChanges[i].Action, gotChanges[i].Action, "change %d", i)
+		assert.Equal(t, wantChanges[i].Kind, gotChanges[i].Kind, "change %d", i)
+		assert.Equal(t, wantChanges[i].Key, gotChanges[i].Key, "change %d", i)
+		assert.Equal(t, wantChanges[i].Version, gotChanges[i].Version, "change %d", i)
+		assert.Equal(t, string(wantChanges[i].Object), string(gotChanges[i].Object), "change %d", i)
+	}
+
+	wantCollections, err := want.Collections()
+	require.NoError(t, err)
+	gotCollections, err := got.Collections()
+	require.NoError(t, err)
+	wantItems := flattenCollectionsForComparison(wantCollections)
+	gotItems := flattenCollectionsForComparison(gotCollections)
+	assert.Equal(t, wantItems, gotItems)
+}
+
+// The scalar event decoders must accept and reject an integer field exactly as decoding the
+// corresponding subsystems type via encoding/json does -- encoding/json is what the
+// launchdarkly_easyjson build uses, so this is the parity contract between the two build variants.
+// The oracle is computed live from encoding/json into the same struct that build decodes into, so
+// per-type behavior is captured automatically: PutObject, DeleteObject, and Payload have plain int
+// fields and reject non-integers, while Selector reads its version through float64 and truncates.
+// Both build variants run this test.
+func TestScalarEventIntegerParsingMatchesEncodingJSON(t *testing.T) {
+	flagJSON := makeTestFlagJSON(t, "flagkey", 1)
+	values := []string{
+		"0", "1", "-1", "100", "2147483647", "-2147483648",
+		"9007199254740993",    // > 2^53, must round-trip exactly for the strict fields
+		"9223372036854775807", // MaxInt64
+		"9223372036854775808", // MaxInt64 + 1 (out of range)
+		"1e20", "-1e20",
+		"1.9", "1.0", "1e2", // non-integer literals
+		"null",
+	}
+
+	t.Run("put-object version", func(t *testing.T) {
+		for _, v := range values {
+			data := []byte(fmt.Sprintf(`{"kind":"flag","key":"k","version":%s,"object":%s}`, v, flagJSON))
+			var oracle subsystems.PutObject
+			wantErr := json.Unmarshal(data, &oracle) != nil
+			p, err := parsePutObjectEventData(data)
+			require.Equalf(t, wantErr, err != nil, "version=%s: encoding/json err=%v, decoder err=%v", v, wantErr, err)
+			if !wantErr {
+				assert.Equalf(t, oracle.Version, p.version, "version=%s", v)
+			}
+		}
+	})
+
+	t.Run("delete-object version", func(t *testing.T) {
+		for _, v := range values {
+			data := []byte(fmt.Sprintf(`{"kind":"flag","key":"k","version":%s}`, v))
+			var oracle subsystems.DeleteObject
+			wantErr := json.Unmarshal(data, &oracle) != nil
+			d, err := parseDeleteObjectEventData(data)
+			require.Equalf(t, wantErr, err != nil, "version=%s: encoding/json err=%v, decoder err=%v", v, wantErr, err)
+			if !wantErr {
+				assert.Equalf(t, oracle.Version, d.Version, "version=%s", v)
+			}
+		}
+	})
+
+	t.Run("server-intent target", func(t *testing.T) {
+		for _, v := range values {
+			data := []byte(fmt.Sprintf(`{"payloads":[{"id":"p1","target":%s,"intentCode":"xfer-full","reason":"r"}]}`, v))
+			var oracle subsystems.ServerIntent
+			wantErr := json.Unmarshal(data, &oracle) != nil
+			intent, err := parseServerIntentEventData(data)
+			require.Equalf(t, wantErr, err != nil, "target=%s: encoding/json err=%v, decoder err=%v", v, wantErr, err)
+			if !wantErr {
+				assert.Equalf(t, oracle.Payload.Target, intent.Payload.Target, "target=%s", v)
+			}
+		}
+	})
+
+	t.Run("selector version", func(t *testing.T) {
+		for _, v := range values {
+			data := []byte(fmt.Sprintf(`{"state":"s","version":%s}`, v))
+			var oracle subsystems.Selector
+			wantErr := json.Unmarshal(data, &oracle) != nil
+			sel, err := parseSelectorEventData(data)
+			require.Equalf(t, wantErr, err != nil, "version=%s: encoding/json err=%v, decoder err=%v", v, wantErr, err)
+			if !wantErr {
+				assert.Equalf(t, oracle.Version(), sel.Version(), "version=%s", v)
+			}
+		}
+	})
+}
+
+// A polling event that carries the "data" property more than once (non-conformant, but handled for
+// parity with encoding/json) resolves last-wins: the event is applied exactly once, using the final
+// value. Both build variants run this test.
+func TestParsePollingPayloadDuplicateDataPropertyUsesLastValue(t *testing.T) {
+	flagJSON := makeTestFlagJSON(t, "flagkey", 1)
+	body := []byte(fmt.Sprintf(`{"events":[`+
+		`{"event":"server-intent","data":{"payloads":[{"id":"p1","target":1,"intentCode":"xfer-full","reason":"r"}]}},`+
+		`{"event":"put-object",`+
+		`"data":{"kind":"flag","key":"first","version":1,"object":%s},`+
+		`"data":{"kind":"flag","key":"last","version":1,"object":%s}},`+
+		`{"event":"payload-transferred","data":{"state":"s","version":1}}`+
+		`]}`, flagJSON, flagJSON))
+
+	changeSet, err := parsePollingPayload(context.Background(), body)
+	require.NoError(t, err)
+
+	changes := changeSet.Changes()
+	require.Len(t, changes, 1)
+	assert.Equal(t, "last", changes[0].Key)
+
+	collections, err := changeSet.Collections()
+	require.NoError(t, err)
+	total := 0
+	for _, coll := range collections {
+		total += len(coll.Items)
+	}
+	assert.Equal(t, 1, total)
+}
+
+func flattenCollectionsForComparison(collections []ldstoretypes.Collection) map[string]ldstoretypes.ItemDescriptor {
+	result := map[string]ldstoretypes.ItemDescriptor{}
+	for _, coll := range collections {
+		for _, item := range coll.Items {
+			result[coll.Kind.GetName()+"/"+item.Key] = item.Item
+		}
+	}
+	return result
+}
+
+// A duplicated "kind" property (non-conformant, but json resolves it last-wins) must not leave the
+// parsed item typed as the earlier kind. The item and the stored kind must always agree: if the
+// final kind is recognized the item is (re-)parsed under it, and if the final kind is unrecognized
+// the object is kept raw with no item. Both build variants run this test.
+func TestParsePutObjectEventDataDuplicateKind(t *testing.T) {
+	flagJSON := makeTestFlagJSON(t, "dup", 5)
+
+	t.Run("later kind wins and item is re-typed", func(t *testing.T) {
+		data := []byte(fmt.Sprintf(`{"kind":"flag","object":%s,"key":"dup","version":5,"kind":"segment"}`, flagJSON))
+		p, err := parsePutObjectEventData(data)
+		require.NoError(t, err)
+		assert.Equal(t, subsystems.SegmentKind, p.kind)
+		require.True(t, p.hasItem)
+		assert.IsType(t, &ldmodel.Segment{}, p.item.Item)
+	})
+
+	t.Run("later unrecognized kind keeps object raw", func(t *testing.T) {
+		data := []byte(fmt.Sprintf(`{"kind":"flag","object":%s,"key":"dup","version":5,"kind":"future-kind"}`, flagJSON))
+		p, err := parsePutObjectEventData(data)
+		require.NoError(t, err)
+		assert.Equal(t, subsystems.ObjectKind("future-kind"), p.kind)
+		assert.False(t, p.hasItem)
+		assert.JSONEq(t, string(flagJSON), string(p.object))
+	})
+}
+
+// A deeply nested value in an unknown/skipped field must be rejected as an error rather than
+// recursing until the stack overflows (a remote-crash DoS from untrusted input). The value exceeds
+// encoding/json's nesting cap; both build variants must return an error, not crash. Both build
+// variants run this test.
+func TestDeeplyNestedSkippedValueIsRejected(t *testing.T) {
+	const depth = 100000
+	nested := strings.Repeat("[", depth) + strings.Repeat("]", depth)
+
+	t.Run("polling top-level unknown property", func(t *testing.T) {
+		body := []byte(`{"junk":` + nested + `,"events":[]}`)
+		_, err := parsePollingPayload(context.Background(), body)
+		assert.Error(t, err)
+	})
+
+	t.Run("put-object unknown property", func(t *testing.T) {
+		data := []byte(`{"kind":"flag","junk":` + nested + `,"key":"k","version":1}`)
+		_, err := parsePutObjectEventData(data)
+		assert.Error(t, err)
+	})
+
+	t.Run("server-intent unknown property", func(t *testing.T) {
+		data := []byte(`{"junk":` + nested + `,"payloads":[{"id":"p1","intentCode":"xfer-full"}]}`)
+		_, err := parseServerIntentEventData(data)
+		assert.Error(t, err)
+	})
+}
diff --git a/internal/datasourcev2/polling_http_request.go b/internal/datasourcev2/polling_http_request.go
index b44554bb..83f49a99 100644
--- a/internal/datasourcev2/polling_http_request.go
+++ b/internal/datasourcev2/polling_http_request.go
@@ -2,7 +2,7 @@ package datasourcev2
 
 import (
 	"context"
-	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"maps"
@@ -83,53 +83,14 @@ func (r *pollingRequester) Request(
 		return subsystems.NewChangeSetBuilder().NoChanges(), headers, nil
 	}
 
-	var payload subsystems.PollingPayload
-	if err = json.Unmarshal(body, &payload); err != nil {
-		return nil, headers, malformedJSONError{err}
-	}
-
-	changeSet := subsystems.NewChangeSetBuilder()
-
-	for _, event := range payload.Events {
-		select {
-		case <-ctx.Done():
-			return nil, headers, ctx.Err()
-		default:
-			switch event.Name {
-			case subsystems.EventServerIntent:
-				var serverIntent subsystems.ServerIntent
-				err := json.Unmarshal(event.Data, &serverIntent)
-				if err != nil {
-					return nil, headers, err
-				}
-				if serverIntent.Payload.Code == subsystems.IntentNone {
-					return changeSet.NoChanges(), headers, nil
-				}
-				changeSet.Start(serverIntent)
-			case subsystems.EventPutObject:
-				var put subsystems.PutObject
-				if err := json.Unmarshal(event.Data, &put); err != nil {
-					return nil, headers, err
-				}
-				changeSet.AddPut(put.Kind, put.Key, put.Version, put.Object)
-			case subsystems.EventDeleteObject:
-				var deleteObject subsystems.DeleteObject
-				if err := json.Unmarshal(event.Data, &deleteObject); err != nil {
-					return nil, headers, err
-				}
-				changeSet.AddDelete(deleteObject.Kind, deleteObject.Key, deleteObject.Version)
-			case subsystems.EventPayloadTransferred:
-				var selector subsystems.Selector
-				if err := json.Unmarshal(event.Data, &selector); err != nil {
-					return nil, headers, err
-				}
-				changeset, err := changeSet.Finish(selector)
-				return changeset, headers, err
-			}
+	changeSet, err := parsePollingPayload(ctx, body)
+	if err != nil {
+		if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+			return nil, headers, err
 		}
+		return nil, headers, malformedJSONError{err}
 	}
-
-	return nil, headers, fmt.Errorf("didn't receive any known protocol events in polling payload")
+	return changeSet, headers, nil
 }
 
 func (r *pollingRequester) makeRequest(
diff --git a/internal/datasourcev2/streaming_data_source.go b/internal/datasourcev2/streaming_data_source.go
index 15f6fe19..d477e125 100644
--- a/internal/datasourcev2/streaming_data_source.go
+++ b/internal/datasourcev2/streaming_data_source.go
@@ -2,7 +2,6 @@ package datasourcev2
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
 	"maps"
 	"net/http"
@@ -214,8 +213,7 @@ func (sp *StreamProcessor) consumeStream(stream *es.Stream, resultChan chan<- su
 				// Swallow the event and move on.
 			case subsystems.EventServerIntent:
 
-				var serverIntent subsystems.ServerIntent
-				err := json.Unmarshal([]byte(event.Data()), &serverIntent)
+				serverIntent, err := parseServerIntentEventData([]byte(event.Data()))
 				if err != nil {
 					gotMalformedEvent(event, err)
 					break
@@ -241,24 +239,21 @@ func (sp *StreamProcessor) consumeStream(stream *es.Stream, resultChan chan<- su
 				}
 
 			case subsystems.EventPutObject:
-				var p subsystems.PutObject
-				err := json.Unmarshal([]byte(event.Data()), &p)
+				p, err := parsePutObjectEventData([]byte(event.Data()))
 				if err != nil {
 					gotMalformedEvent(event, err)
 					break
 				}
-				changeSetBuilder.AddPut(p.Kind, p.Key, p.Version, p.Object)
+				p.addTo(changeSetBuilder)
 			case subsystems.EventDeleteObject:
-				var d subsystems.DeleteObject
-				err := json.Unmarshal([]byte(event.Data()), &d)
+				d, err := parseDeleteObjectEventData([]byte(event.Data()))
 				if err != nil {
 					gotMalformedEvent(event, err)
 					break
 				}
 				changeSetBuilder.AddDelete(d.Kind, d.Key, d.Version)
 			case subsystems.EventGoodbye:
-				var goodbye subsystems.Goodbye
-				err := json.Unmarshal([]byte(event.Data()), &goodbye)
+				goodbye, err := parseGoodbyeEventData([]byte(event.Data()))
 				if err != nil {
 					gotMalformedEvent(event, err)
 					break
@@ -266,8 +261,7 @@ func (sp *StreamProcessor) consumeStream(stream *es.Stream, resultChan chan<- su
 
 				sp.loggers.Infof("SSE server sent goodbye: %s", goodbye.Reason)
 			case subsystems.EventError:
-				var errorData subsystems.Error
-				err := json.Unmarshal([]byte(event.Data()), &errorData)
+				errorData, err := parseErrorEventData([]byte(event.Data()))
 				if err != nil {
 					gotMalformedEvent(event, err)
 					break
@@ -284,8 +278,7 @@ func (sp *StreamProcessor) consumeStream(stream *es.Stream, resultChan chan<- su
 				changeSetBuilder.Reset()
 
 			case subsystems.EventPayloadTransferred:
-				var selector subsystems.Selector
-				err := json.Unmarshal([]byte(event.Data()), &selector)
+				selector, err := parseSelectorEventData([]byte(event.Data()))
 				if err != nil {
 					gotMalformedEvent(event, err)
 					break
diff --git a/subsystems/changeset.go b/subsystems/changeset.go
index ed294b10..94bfe7c3 100644
--- a/subsystems/changeset.go
+++ b/subsystems/changeset.go
@@ -122,7 +122,17 @@ func toStorableItems(deltas []Change) ([]ldstoretypes.Collection, error) {
 // ChangeSetBuilder is a helper for constructing a ChangeSet.
 type ChangeSetBuilder struct {
 	intent  *ServerIntent
-	changes []Change
+	changes []changeRecord
+}
+
+// changeRecord is a Change plus, optionally, the already-parsed representation of the change's
+// object. When every put of a recognized kind carries a parsed item, the builder can assemble
+// the ChangeSet's collections directly at Finish time, so that Collections() does not need to
+// re-parse the raw JSON.
+type changeRecord struct {
+	change  Change
+	item    ldstoretypes.ItemDescriptor
+	hasItem bool
 }
 
 // NewChangeSetBuilder creates a new ChangeSetBuilder, which is empty by default.
@@ -265,12 +275,22 @@ func (c *ChangeSetBuilder) Finish(selector Selector) (*ChangeSet, error) {
 	if c.intent == nil {
 		return nil, errors.New("changeset: cannot complete without a server-intent")
 	}
-	changes := &ChangeSet{
+	var changes []Change
+	if c.changes != nil {
+		changes = make([]Change, 0, len(c.changes))
+		for _, record := range c.changes {
+			changes = append(changes, record.change)
+		}
+	}
+	changeSet := &ChangeSet{
 		intentCode: c.intent.Payload.Code,
 		selector:   selector,
-		changes:    c.changes,
+		changes:    changes,
 		mu:         &sync.Mutex{},
 	}
+	if collections, ok := collectionsFromRecords(c.changes); ok {
+		changeSet.collection = collections
+	}
 	c.changes = nil
 
 	// Once a full transfer has been processed, all future changes should be
@@ -279,17 +299,91 @@ func (c *ChangeSetBuilder) Finish(selector Selector) (*ChangeSet, error) {
 	if c.intent.Payload.Code == IntentTransferFull {
 		c.intent.Payload.Code = IntentTransferChanges
 	}
-	return changes, nil
+	return changeSet, nil
+}
+
+// collectionsFromRecords assembles storable collections from the builder's records without
+// parsing any JSON, if possible. This requires every put of a recognized kind to carry a
+// pre-parsed item; otherwise it returns (nil, false) and the ChangeSet falls back to parsing
+// lazily in Collections().
+func collectionsFromRecords(records []changeRecord) ([]ldstoretypes.Collection, bool) {
+	if records == nil {
+		return nil, false
+	}
+	for _, record := range records {
+		if record.change.Action == ChangeTypePut && !record.hasItem {
+			if _, recognized := record.change.Kind.ToFDV1(); recognized {
+				return nil, false
+			}
+		}
+	}
+	collections := make(kindMap)
+	for _, record := range records {
+		kind, ok := record.change.Kind.ToFDV1()
+		if !ok {
+			// Unrecognized kinds are not errors and are ignored for forwards compatibility.
+			continue
+		}
+		switch record.change.Action {
+		case ChangeTypePut:
+			collections[kind] = append(collections[kind], ldstoretypes.KeyedItemDescriptor{
+				Key:  record.change.Key,
+				Item: record.item,
+			})
+		case ChangeTypeDelete:
+			// A deletion is represented by a tombstone, which is an ItemDescriptor with a version and nil item.
+			collections[kind] = append(collections[kind], ldstoretypes.KeyedItemDescriptor{
+				Key:  record.change.Key,
+				Item: ldstoretypes.ItemDescriptor{Version: record.change.Version, Item: nil},
+			})
+		default:
+			// An unknown action isn't an error, and should be ignored for forwards compatibility.
+			continue
+		}
+	}
+	result := collections.flatten()
+	if result == nil {
+		result = []ldstoretypes.Collection{}
+	}
+	return result, true
 }
 
 // AddPut adds a new object to the changeset.
 func (c *ChangeSetBuilder) AddPut(kind ObjectKind, key string, version int, object json.RawMessage) *ChangeSetBuilder {
-	c.changes = append(c.changes, Change{
+	c.changes = append(c.changes, changeRecord{change: Change{
 		Action:  ChangeTypePut,
 		Kind:    kind,
 		Key:     key,
 		Version: version,
 		Object:  object,
+	}})
+
+	return c
+}
+
+// AddParsedPut adds a new object to the changeset, along with the already-parsed representation
+// of that object. The object parameter is the raw JSON of the item, exactly as it would be passed
+// to AddPut, and item is the result of deserializing those same bytes.
+//
+// Providing the parsed item up front allows the ChangeSet returned by Finish to assemble its
+// Collections() result directly, instead of re-parsing the raw JSON of every put.
+func (c *ChangeSetBuilder) AddParsedPut(
+	kind ObjectKind,
+	key string,
+	version int,
+	object json.RawMessage,
+	item ldstoretypes.ItemDescriptor,
+) *ChangeSetBuilder {
+	c.changes = append(c.changes, changeRecord{
+		change: Change{
+			Action:  ChangeTypePut,
+			Kind:    kind,
+			Key:     key,
+			Version: version,
+			Object:  object,
+		},
+		item:    item,
+		hasItem: true,
 	})
 
 	return c
@@ -297,12 +391,12 @@ func (c *ChangeSetBuilder) AddPut(kind ObjectKind, key string, version int, obje
 
 // AddDelete adds a deletion to the changeset.
 func (c *ChangeSetBuilder) AddDelete(kind ObjectKind, key string, version int) *ChangeSetBuilder {
-	c.changes = append(c.changes, Change{
+	c.changes = append(c.changes, changeRecord{change: Change{
 		Action:  ChangeTypeDelete,
 		Kind:    kind,
 		Key:     key,
 		Version: version,
-	})
+	}})
 
 	return c
 }
diff --git a/subsystems/changeset_parsed_put_test.go b/subsystems/changeset_parsed_put_test.go
new file mode 100644
index 00000000..73afde6a
--- /dev/null
+++ b/subsystems/changeset_parsed_put_test.go
@@ -0,0 +1,133 @@
+package subsystems
+
+import (
+	"testing"
+
+	"github.com/launchdarkly/go-sdk-common/v3/ldvalue"
+	"github.com/launchdarkly/go-server-sdk-evaluation/v3/ldbuilders"
+	"github.com/launchdarkly/go-server-sdk-evaluation/v3/ldmodel"
+	"github.com/launchdarkly/go-server-sdk/v7/internal/datakinds"
+	"github.com/launchdarkly/go-server-sdk/v7/subsystems/ldstoretypes"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func makeParsedFlagItem(key string, version int) ldstoretypes.ItemDescriptor {
+	flag := ldbuilders.NewFlagBuilder(key).Version(version).SingleVariation(ldvalue.Bool(true)).Build()
+	return ldstoretypes.ItemDescriptor{Version: version, Item: &flag}
+}
+
+func serializeFlagItem(item ldstoretypes.ItemDescriptor) []byte {
+	return datakinds.Features.Serialize(item)
+}
+
+func TestChangeSetBuilderAddParsedPutPrePopulatesCollections(t *testing.T) {
+	builder := NewChangeSetBuilder()
+	builder.Start(ServerIntent{Payload: Payload{Code: IntentTransferFull}})
+
+	// The raw object bytes are deliberately NOT valid JSON. If Collections() were to re-parse
+	// them, it would fail; succeeding proves the pre-parsed items were used instead.
+	builder.AddParsedPut(FlagKind, "flag-1", 1, []byte("deliberately not JSON"), makeParsedFlagItem("flag-1", 1))
+	builder.AddDelete(FlagKind, "flag-2", 2)
+
+	changeSet, err := builder.Finish(NewSelector("state", 1))
+	require.NoError(t, err)
+
+	// The raw bytes are preserved verbatim for consumers of Changes().
+	changes := changeSet.Changes()
+	require.Len(t, changes, 2)
+	assert.Equal(t, "deliberately not JSON", string(changes[0].Object))
+
+	collections, err := changeSet.Collections()
+	require.NoError(t, err)
+	require.Len(t, collections, 1)
+	assert.Equal(t, "features", collections[0].Kind.GetName())
+	require.Len(t, collections[0].Items, 2)
+
+	byKey := map[string]ldstoretypes.ItemDescriptor{}
+	for _, item := range collections[0].Items {
+		byKey[item.Key] = item.Item
+	}
+	require.IsType(t, &ldmodel.FeatureFlag{}, byKey["flag-1"].Item)
+	assert.Equal(t, 1, byKey["flag-1"].Version)
+	assert.Nil(t, byKey["flag-2"].Item) // the delete is a tombstone
+	assert.Equal(t, 2, byKey["flag-2"].Version)
+}
+
+func TestChangeSetBuilderMixedPutsFallBackToLazyParsing(t *testing.T) {
+	builder := NewChangeSetBuilder()
+	builder.Start(ServerIntent{Payload: Payload{Code: IntentTransferFull}})
+
+	item1 := makeParsedFlagItem("flag-1", 1)
+	builder.AddParsedPut(FlagKind, "flag-1", 1, serializeFlagItem(item1), item1)
+	// This put has no parsed item, so the whole change set must be parsed lazily.
+	builder.AddPut(FlagKind, "flag-2", 2, serializeFlagItem(makeParsedFlagItem("flag-2", 2)))
+
+	changeSet, err := builder.Finish(NewSelector("state", 1))
+	require.NoError(t, err)
+
+	collections, err := changeSet.Collections()
+	require.NoError(t, err)
+	require.Len(t, collections, 1)
+	assert.Len(t, collections[0].Items, 2)
+	for _, item := range collections[0].Items {
+		assert.NotNil(t, item.Item.Item, "item %s should have been parsed", item.Key)
+	}
+}
+
+func TestChangeSetBuilderUnparsedPutOfUnknownKindDoesNotBlockPrePopulation(t *testing.T) {
+	builder := NewChangeSetBuilder()
+	builder.Start(ServerIntent{Payload: Payload{Code: IntentTransferFull}})
+
+	// An unrecognized kind never has a parsed item, and is excluded from collections; its
+	// presence must not force the lazy path (which the invalid raw bytes of the parsed put
+	// would then fail on).
+	builder.AddPut("future-kind", "future-key", 1, []byte(`{"anything":true}`))
+	builder.AddParsedPut(FlagKind, "flag-1", 1, []byte("deliberately not JSON"), makeParsedFlagItem("flag-1", 1))
+
+	changeSet, err := builder.Finish(NewSelector("state", 1))
+	require.NoError(t, err)
+
+	// Both changes are visible to Changes() consumers...
+	require.Len(t, changeSet.Changes(), 2)
+	assert.Equal(t, ObjectKind("future-kind"), changeSet.Changes()[0].Kind)
+
+	// ...but only the recognized kind appears in collections.
+	collections, err := changeSet.Collections()
+	require.NoError(t, err)
+	require.Len(t, collections, 1)
+	assert.Equal(t, "features", collections[0].Kind.GetName())
+	assert.Len(t, collections[0].Items, 1)
+}
+
+func TestChangeSetBuilderParsedPutsAcrossFinishReuse(t *testing.T) {
+	builder := NewChangeSetBuilder()
+	builder.Start(ServerIntent{Payload: Payload{Code: IntentTransferFull}})
+	builder.AddParsedPut(FlagKind, "flag-1", 1, []byte("raw-1"), makeParsedFlagItem("flag-1", 1))
+	first, err := builder.Finish(NewSelector("state", 1))
+	require.NoError(t, err)
+
+	// The builder is reusable after Finish; the next change set starts empty and the intent
+	// becomes xfer-changes.
+	builder.AddParsedPut(FlagKind, "flag-2", 2, []byte("raw-2"), makeParsedFlagItem("flag-2", 2))
+	second, err := builder.Finish(NewSelector("state", 2))
+	require.NoError(t, err)
+
+	require.Len(t, first.Changes(), 1)
+	assert.Equal(t, "flag-1", first.Changes()[0].Key)
+	assert.Equal(t, IntentTransferFull, first.IntentCode())
+
+	require.Len(t, second.Changes(), 1)
+	assert.Equal(t, "flag-2", second.Changes()[0].Key)
+	assert.Equal(t, IntentTransferChanges, second.IntentCode())
+
+	firstCollections, err := first.Collections()
+	require.NoError(t, err)
+	secondCollections, err := second.Collections()
+	require.NoError(t, err)
+	require.Len(t, firstCollections, 1)
+	require.Len(t, secondCollections, 1)
+	assert.Equal(t, "flag-1", firstCollections[0].Items[0].Key)
+	assert.Equal(t, "flag-2", secondCollections[0].Items[0].Key)
+}