diff --git a/datatrack/clock.go b/datatrack/clock.go
new file mode 100644
index 00000000..67446d83
--- /dev/null
+++ b/datatrack/clock.go
@@ -0,0 +1,77 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package datatrack
+
+import (
+	"math/rand/v2"
+	"time"
+)
+
+const clockRate = 90_000
+
+// timestamp is a packet-level timestamp in ticks of clockRate.
+type timestamp uint32
+
+func randomTimestamp() timestamp {
+	return timestamp(rand.Uint32())
+}
+
+func (t timestamp) isBefore(other timestamp) bool {
+	if t-other == 1<<31 {
+		return t < other
+	}
+	return int32(t-other) < 0
+}
+
+func (t timestamp) wrappingAdd(ticks uint32) timestamp {
+	return timestamp(uint32(t) + ticks)
+}
+
+// clock maps instants to monotonic packet timestamps.
+type clock struct {
+	epoch time.Time
+	base  timestamp
+	prev  timestamp
+}
+
+func newClock(base timestamp) *clock {
+	return newClockWithEpoch(time.Now(), base)
+}
+
+func newClockWithEpoch(epoch time.Time, base timestamp) *clock {
+	return &clock{epoch: epoch, base: base, prev: base}
+}
+
+func (c *clock) now() timestamp {
+	return c.at(time.Now())
+}
+
+func (c *clock) at(instant time.Time) timestamp {
+	ts := c.base.wrappingAdd(durationToTicks(instant.Sub(c.epoch)))
+	if ts.isBefore(c.prev) {
+		ts = c.prev
+	}
+	c.prev = ts
+	return ts
+}
+
+func durationToTicks(d time.Duration) uint32 {
+	nanos := d.Nanoseconds()
+	if nanos < 0 {
+		nanos = 0
+	}
+	seconds, remainder := uint64(nanos)/1_000_000_000, uint64(nanos)%1_000_000_000
+	return uint32(seconds*clockRate + (remainder*clockRate+500_000_000)/1_000_000_000)
+}
diff --git a/datatrack/depacketizer.go b/datatrack/depacketizer.go
new file mode 100644
index 00000000..89a02850
--- /dev/null
+++ b/datatrack/depacketizer.go
@@ -0,0 +1,212 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package datatrack
+
+import (
+	"fmt"
+	"slices"
+
+	dtp "github.com/livekit/protocol/datatrack"
+)
+
+const maxBufferPackets = 128
+
+// depacketizer reassembles packets into frames.
+type depacketizer struct {
+	partials map[uint16]*partialFrame
+	// frame numbers of partials in insertion order, oldest first
+	order []uint16
+}
+
+type partialFrame struct {
+	startSequence uint16
+	extensions    Extensions
+	payloads      map[uint16][]byte
+}
+
+type depacketizerFrame struct {
+	payload    []byte
+	extensions Extensions
+}
+
+type depacketizerPushOptions struct {
+	// maxPartialFrames is the number of frames assembled concurrently; the oldest partial is
+	// evicted when a new frame arrives at capacity.
+	maxPartialFrames int
+}
+
+// depacketizerPushResult may carry both a completed frame and a dropped one, since a single
+// packet can complete one frame while evicting another.
+type depacketizerPushResult struct {
+	frame *depacketizerFrame
+	drop  *depacketizerDropError
+}
+
+type depacketizerDropReason int
+
+const (
+	depacketizerDropReasonInterrupted depacketizerDropReason = iota
+	depacketizerDropReasonUnknownFrame
+	depacketizerDropReasonBufferFull
+	depacketizerDropReasonIncomplete
+)
+
+// depacketizerDropError records a dropped frame.
+type depacketizerDropError struct {
+	frameNumber uint16
+	reason      depacketizerDropReason
+	// newFrameNumber is the frame that interrupted the dropped one
+	newFrameNumber uint16
+	// received and expected packet counts of an incomplete frame
+	received, expected uint16
+}
+
+func (e *depacketizerDropError) Error() string {
+	var reason string
+	switch e.reason {
+	case depacketizerDropReasonInterrupted:
+		reason = fmt.Sprintf("interrupted by new frame %d", e.newFrameNumber)
+	case depacketizerDropReasonUnknownFrame:
+		reason = "unknown frame"
+	case depacketizerDropReasonBufferFull:
+		reason = "buffer full"
+	case depacketizerDropReasonIncomplete:
+		reason = fmt.Sprintf("incomplete (%d/%d)", e.received, e.expected)
+	}
+	return fmt.Sprintf("frame %d dropped: %s", e.frameNumber, reason)
+}
+
+func newDepacketizer() *depacketizer {
+	return &depacketizer{partials: make(map[uint16]*partialFrame)}
+}
+
+func (d *depacketizer) push(packet dtp.Packet, options depacketizerPushOptions) depacketizerPushResult {
+	switch markerOf(&packet.Header) {
+	case FrameMarkerSingle:
+		return d.frameFromSingle(packet, options)
+	case FrameMarkerStart:
+		return d.beginPartial(packet, options)
+	default:
+		return d.pushToPartial(packet)
+	}
+}
+
+func (d *depacketizer) frameFromSingle(packet dtp.Packet, options depacketizerPushOptions) depacketizerPushResult {
+	var result depacketizerPushResult
+	// A single packet is a self-contained frame and does not occupy a partials slot, but if
+	// the partials map is at capacity it is treated as a signal that the oldest in-flight
+	// partial is stale and evicts it.
+	if len(d.partials) >= options.maxPartialFrames {
+		result.drop = d.evictOldest(packet.FrameNumber)
+	}
+	result.frame = &depacketizerFrame{payload: packet.Payload, extensions: extensionsOf(&packet.Header)}
+	return result
+}
+
+func (d *depacketizer) evictOldest(newFrameNumber uint16) *depacketizerDropError {
+	if len(d.order) == 0 {
+		return nil
+	}
+	oldest := d.order[0]
+	d.remove(oldest)
+	return &depacketizerDropError{frameNumber: oldest, reason: depacketizerDropReasonInterrupted, newFrameNumber: newFrameNumber}
+}
+
+func (d *depacketizer) remove(frameNumber uint16) {
+	delete(d.partials, frameNumber)
+	if i := slices.Index(d.order, frameNumber); i >= 0 {
+		d.order = slices.Delete(d.order, i, i+1)
+	}
+}
+
+func (d *depacketizer) beginPartial(packet dtp.Packet, options depacketizerPushOptions) depacketizerPushResult {
+	var result depacketizerPushResult
+	for len(d.partials) >= options.maxPartialFrames {
+		evicted := d.evictOldest(packet.FrameNumber)
+		if evicted == nil {
+			break
+		}
+		if result.drop == nil {
+			result.drop = evicted
+		}
+	}
+
+	if _, replacing := d.partials[packet.FrameNumber]; !replacing {
+		d.order = append(d.order, packet.FrameNumber)
+	}
+	d.partials[packet.FrameNumber] = &partialFrame{
+		startSequence: packet.SequenceNumber,
+		extensions:    extensionsOf(&packet.Header),
+		payloads:      map[uint16][]byte{packet.SequenceNumber: packet.Payload},
+	}
+	return result
+}
+
+func (d *depacketizer) pushToPartial(packet dtp.Packet) depacketizerPushResult {
+	frameNumber := packet.FrameNumber
+	partial, ok := d.partials[frameNumber]
+	if !ok {
+		return depacketizerPushResult{drop: &depacketizerDropError{frameNumber: frameNumber, reason: depacketizerDropReasonUnknownFrame}}
+	}
+
+	if len(partial.payloads) >= maxBufferPackets {
+		d.remove(frameNumber)
+		return depacketizerPushResult{drop: &depacketizerDropError{frameNumber: frameNumber, reason: depacketizerDropReasonBufferFull}}
+	}
+
+	partial.payloads[packet.SequenceNumber] = packet.Payload
+
+	if markerOf(&packet.Header) == FrameMarkerFinal {
+		d.remove(frameNumber)
+		return finalize(frameNumber, partial, packet.SequenceNumber)
+	}
+	return depacketizerPushResult{}
+}
+
+func finalize(frameNumber uint16, partial *partialFrame, endSequence uint16) depacketizerPushResult {
+	received := uint16(len(partial.payloads))
+	expected := endSequence - partial.startSequence + 1
+	if received != expected {
+		return depacketizerPushResult{drop: &depacketizerDropError{
+			frameNumber: frameNumber,
+			reason:      depacketizerDropReasonIncomplete,
+			received:    received,
+			expected:    expected,
+		}}
+	}
+
+	payloadLen := 0
+	for _, payload := range partial.payloads {
+		payloadLen += len(payload)
+	}
+	payload := make([]byte, 0, payloadLen)
+
+	for sequence := partial.startSequence; ; sequence++ {
+		partialPayload, ok := partial.payloads[sequence]
+		if !ok {
+			break
+		}
+		payload = append(payload, partialPayload...)
+		if sequence == endSequence {
+			return depacketizerPushResult{frame: &depacketizerFrame{payload: payload, extensions: partial.extensions}}
+		}
+	}
+	return depacketizerPushResult{drop: &depacketizerDropError{
+		frameNumber: frameNumber,
+		reason:      depacketizerDropReasonIncomplete,
+		received:    received,
+		expected:    expected,
+	}}
+}
diff --git a/datatrack/depacketizer_test.go b/datatrack/depacketizer_test.go
new file mode 100644
index 00000000..7488d779
--- /dev/null
+++ b/datatrack/depacketizer_test.go
@@ -0,0 +1,433 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package datatrack
+
+import (
+	"bytes"
+	"fmt"
+	"testing"
+
+	dtp "github.com/livekit/protocol/datatrack"
+	"github.com/stretchr/testify/require"
+)
+
+var testDefaultPushOptions = depacketizerPushOptions{maxPartialFrames: 1}
+
+// derive returns a copy of base with the given marker, sequence and frame number.
+func derive(base dtp.Packet, marker FrameMarker, sequence, frameNumber uint16) dtp.Packet {
+	packet := base
+	marker.apply(&packet.Header)
+	packet.SequenceNumber = sequence
+	packet.FrameNumber = frameNumber
+	return packet
+}
+
+func withPayload(packet dtp.Packet, payload []byte) dtp.Packet {
+	packet.Payload = payload
+	return packet
+}
+
+func TestDepacketizer_SinglePacket(t *testing.T) {
+	d := newDepacketizer()
+
+	packet := testPacket()
+	FrameMarkerSingle.apply(&packet.Header)
+
+	result := d.push(packet, testDefaultPushOptions)
+	require.Nil(t, result.drop)
+	require.NotNil(t, result.frame)
+	require.Equal(t, packet.Payload, result.frame.payload)
+	require.Equal(t, extensionsOf(&packet.Header), result.frame.extensions)
+}
+
+func TestDepacketizer_MultiPacket(t *testing.T) {
+	for _, tc := range []struct {
+		name         string
+		interPackets int
+	}{
+		{"0", 0},
+		{"8", 8},
+		{"buffer_limit", maxBufferPackets - 2},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			d := newDepacketizer()
+
+			packet := testPacket()
+			FrameMarkerStart.apply(&packet.Header)
+
+			result := d.push(packet, testDefaultPushOptions)
+			require.True(t, result.frame == nil && result.drop == nil)
+
+			for range tc.interPackets {
+				FrameMarkerInter.apply(&packet.Header)
+				packet.SequenceNumber++
+
+				result = d.push(packet, testDefaultPushOptions)
+				require.True(t, result.frame == nil && result.drop == nil)
+			}
+
+			FrameMarkerFinal.apply(&packet.Header)
+			packet.SequenceNumber++
+
+			result = d.push(packet, testDefaultPushOptions)
+			require.Nil(t, result.drop)
+			require.NotNil(t, result.frame)
+			require.Equal(t, extensionsOf(&packet.Header), result.frame.extensions)
+			require.Len(t, result.frame.payload, len(packet.Payload)*(tc.interPackets+2))
+		})
+	}
+}
+
+func TestDepacketizer_Interrupted(t *testing.T) {
+	d := newDepacketizer()
+
+	packet := testPacket()
+	FrameMarkerStart.apply(&packet.Header)
+
+	result := d.push(packet, testDefaultPushOptions)
+	require.True(t, result.frame == nil && result.drop == nil)
+
+	firstFrameNumber := packet.FrameNumber
+	newFrameNumber := packet.FrameNumber + 1
+	packet.FrameNumber = newFrameNumber
+
+	result = d.push(packet, testDefaultPushOptions)
+	require.Nil(t, result.frame)
+	require.Equal(t, &depacketizerDropError{
+		frameNumber:    firstFrameNumber,
+		reason:         depacketizerDropReasonInterrupted,
+		newFrameNumber: newFrameNumber,
+	}, result.drop)
+}
+
+func TestDepacketizer_Incomplete(t *testing.T) {
+	d := newDepacketizer()
+
+	packet := testPacket()
+	frameNumber := packet.FrameNumber
+	FrameMarkerStart.apply(&packet.Header)
+
+	d.push(packet, testDefaultPushOptions)
+
+	packet.SequenceNumber += 3
+	FrameMarkerFinal.apply(&packet.Header)
+
+	result := d.push(packet, testDefaultPushOptions)
+	require.Nil(t, result.frame)
+	require.Equal(t, &depacketizerDropError{
+		frameNumber: frameNumber,
+		reason:      depacketizerDropReasonIncomplete,
+		received:    2,
+		expected:    4,
+	}, result.drop)
+}
+
+func TestDepacketizer_UnknownFrame(t *testing.T) {
+	d := newDepacketizer()
+
+	packet := testPacket()
+	frameNumber := packet.FrameNumber
+	FrameMarkerInter.apply(&packet.Header)
+
+	result := d.push(packet, testDefaultPushOptions)
+	require.Equal(t, &depacketizerDropError{frameNumber: frameNumber, reason: depacketizerDropReasonUnknownFrame}, result.drop)
+}
+
+func TestDepacketizer_MultiFrame(t *testing.T) {
+	d := newDepacketizer()
+
+	var sequence uint16
+	next := func() uint16 {
+		current := sequence
+		sequence++
+		return current
+	}
+	for frameNumber := range uint16(10) {
+		packet := testPacket()
+		packet.FrameNumber = frameNumber
+
+		FrameMarkerStart.apply(&packet.Header)
+		packet.SequenceNumber = next()
+		result := d.push(packet, testDefaultPushOptions)
+		require.True(t, result.drop == nil && result.frame == nil)
+
+		FrameMarkerInter.apply(&packet.Header)
+		packet.SequenceNumber = next()
+		result = d.push(packet, testDefaultPushOptions)
+		require.True(t, result.drop == nil && result.frame == nil)
+
+		FrameMarkerFinal.apply(&packet.Header)
+		packet.SequenceNumber = next()
+		result = d.push(packet, testDefaultPushOptions)
+		require.True(t, result.drop == nil && result.frame != nil)
+	}
+}
+
+func TestDepacketizer_DuplicateSequenceNumbers(t *testing.T) {
+	d := newDepacketizer()
+
+	packet := testPacket()
+	FrameMarkerStart.apply(&packet.Header)
+	packet.SequenceNumber = 1
+	packet.Payload = bytes.Repeat([]byte{0xab}, 3)
+
+	result := d.push(packet, testDefaultPushOptions)
+	require.True(t, result.drop == nil && result.frame == nil)
+
+	FrameMarkerInter.apply(&packet.Header)
+	packet.SequenceNumber = 1 // same sequence number
+	packet.Payload = bytes.Repeat([]byte{0xcd}, 3)
+
+	result = d.push(packet, testDefaultPushOptions)
+	require.True(t, result.drop == nil && result.frame == nil)
+
+	FrameMarkerFinal.apply(&packet.Header)
+	packet.SequenceNumber = 2
+	packet.Payload = bytes.Repeat([]byte{0xef}, 3)
+
+	result = d.push(packet, testDefaultPushOptions)
+	require.Nil(t, result.drop)
+	require.NotNil(t, result.frame)
+	require.True(t, bytes.HasPrefix(result.frame.payload, bytes.Repeat([]byte{0xcd}, 3)))
+}
+
+func TestDepacketizer_AssemblesMultiplePartialFrames(t *testing.T) {
+	d := newDepacketizer()
+	opts := depacketizerPushOptions{maxPartialFrames: 2}
+
+	base := testPacket()
+	payloadLen := len(base.Payload)
+
+	result := d.push(derive(base, FrameMarkerStart, 0, 1), opts)
+	require.True(t, result.frame == nil && result.drop == nil)
+
+	result = d.push(derive(base, FrameMarkerStart, 100, 2), opts)
+	require.True(t, result.frame == nil && result.drop == nil)
+
+	result = d.push(derive(base, FrameMarkerFinal, 1, 1), opts)
+	require.Nil(t, result.drop)
+	require.NotNil(t, result.frame)
+	require.Len(t, result.frame.payload, payloadLen*2)
+
+	result = d.push(derive(base, FrameMarkerFinal, 101, 2), opts)
+	require.Nil(t, result.drop)
+	require.NotNil(t, result.frame)
+	require.Len(t, result.frame.payload, payloadLen*2)
+}
+
+func TestDepacketizer_StartingNewFrameAtCapacity(t *testing.T) {
+	d := newDepacketizer()
+	opts := depacketizerPushOptions{maxPartialFrames: 2}
+
+	base := testPacket()
+
+	result := d.push(derive(base, FrameMarkerStart, 0, 1), opts)
+	require.True(t, result.frame == nil && result.drop == nil)
+
+	result = d.push(derive(base, FrameMarkerStart, 100, 2), opts)
+	require.True(t, result.frame == nil && result.drop == nil)
+
+	result = d.push(derive(base, FrameMarkerStart, 200, 3), opts)
+	require.Nil(t, result.frame)
+	require.Equal(t, &depacketizerDropError{frameNumber: 1, reason: depacketizerDropReasonInterrupted, newFrameNumber: 3}, result.drop)
+}
+
+func TestDepacketizer_SinglePacketAtCapacity(t *testing.T) {
+	d := newDepacketizer()
+	opts := depacketizerPushOptions{maxPartialFrames: 2}
+
+	base := testPacket()
+
+	result := d.push(derive(base, FrameMarkerStart, 0, 1), opts)
+	require.True(t, result.frame == nil && result.drop == nil)
+
+	result = d.push(derive(base, FrameMarkerStart, 100, 2), opts)
+	require.True(t, result.frame == nil && result.drop == nil)
+
+	result = d.push(derive(base, FrameMarkerSingle, 200, 3), opts)
+	require.Equal(t, &depacketizerDropError{frameNumber: 1, reason: depacketizerDropReasonInterrupted, newFrameNumber: 3}, result.drop)
+}
+
+func TestDepacketizer_RepeatedStartBelowCapacity(t *testing.T) {
+	d := newDepacketizer()
+	opts := depacketizerPushOptions{maxPartialFrames: 2}
+
+	base := testPacket()
+
+	result := d.push(derive(base, FrameMarkerStart, 0, 1), opts)
+	require.True(t, result.frame == nil && result.drop == nil)
+
+	// a repeated start replaces frame 1 in place
+	result = d.push(derive(base, FrameMarkerStart, 10, 1), opts)
+	require.True(t, result.frame == nil && result.drop == nil)
+	require.Len(t, d.order, len(d.partials))
+
+	result = d.push(derive(base, FrameMarkerStart, 100, 2), opts)
+	require.True(t, result.frame == nil && result.drop == nil)
+
+	result = d.push(derive(base, FrameMarkerStart, 200, 3), opts)
+	require.Equal(t, &depacketizerDropError{frameNumber: 1, reason: depacketizerDropReasonInterrupted, newFrameNumber: 3}, result.drop)
+
+	result = d.push(derive(base, FrameMarkerStart, 300, 4), opts)
+	require.Equal(t, &depacketizerDropError{frameNumber: 2, reason: depacketizerDropReasonInterrupted, newFrameNumber: 4}, result.drop)
+
+	result = d.push(derive(base, FrameMarkerFinal, 301, 4), opts)
+	require.Nil(t, result.drop)
+	require.NotNil(t, result.frame)
+	require.Len(t, result.frame.payload, len(base.Payload)*2)
+}
+
+func TestDepacketizer_EvictsOldestWhenStartsExceedMax(t *testing.T) {
+	d := newDepacketizer()
+	opts := depacketizerPushOptions{maxPartialFrames: 5}
+
+	const totalFrames uint16 = 10
+	base := testPacket()
+
+	for i := range totalFrames {
+		require.Nil(t, d.push(derive(base, FrameMarkerStart, i*2, i+1), opts).frame)
+	}
+
+	producedFrames, unknownFrameErrors := 0, 0
+	for i := range totalFrames {
+		result := d.push(derive(base, FrameMarkerFinal, i*2+1, i+1), opts)
+		if result.frame != nil {
+			producedFrames++
+		}
+		if result.drop != nil {
+			require.Equal(t, depacketizerDropReasonUnknownFrame, result.drop.reason)
+			unknownFrameErrors++
+		}
+	}
+
+	require.Equal(t, 5, producedFrames)
+	require.Equal(t, 5, unknownFrameErrors)
+}
+
+func TestDepacketizer_LatePacketsForEvictedFrame(t *testing.T) {
+	d := newDepacketizer()
+	opts := depacketizerPushOptions{maxPartialFrames: 3}
+
+	base := testPacket()
+
+	for i := uint16(1); i <= 3; i++ {
+		require.Nil(t, d.push(derive(base, FrameMarkerStart, i*100, i), opts).frame)
+	}
+
+	// a fourth start evicts the oldest (frame 1)
+	require.Nil(t, d.push(derive(base, FrameMarkerStart, 400, 4), opts).frame)
+
+	result := d.push(derive(base, FrameMarkerInter, 101, 1), opts)
+	require.Nil(t, result.frame)
+	require.Equal(t, &depacketizerDropError{frameNumber: 1, reason: depacketizerDropReasonUnknownFrame}, result.drop)
+
+	result = d.push(derive(base, FrameMarkerFinal, 102, 1), opts)
+	require.Nil(t, result.frame)
+	require.Equal(t, &depacketizerDropError{frameNumber: 1, reason: depacketizerDropReasonUnknownFrame}, result.drop)
+
+	for _, frameNumber := range []uint16{2, 3, 4} {
+		require.NotNil(t, d.push(derive(base, FrameMarkerFinal, frameNumber*100+1, frameNumber), opts).frame)
+	}
+}
+
+func TestDepacketizer_HeavilyInterleavedFrames(t *testing.T) {
+	d := newDepacketizer()
+	opts := depacketizerPushOptions{maxPartialFrames: 3}
+
+	base := testPacket()
+
+	type frameSpec struct {
+		frameNumber   uint16
+		startSequence uint16
+		payloads      [3][]byte
+	}
+	frames := []frameSpec{
+		{1, 0, [3][]byte{{0xa1}, {0xa2}, {0xa3}}},
+		{2, 100, [3][]byte{{0xb1}, {0xb2}, {0xb3}}},
+		{3, 200, [3][]byte{{0xc1}, {0xc2}, {0xc3}}},
+	}
+	build := func(frameIdx int, packetIdx uint16, marker FrameMarker) dtp.Packet {
+		f := frames[frameIdx]
+		return withPayload(derive(base, marker, f.startSequence+packetIdx, f.frameNumber), f.payloads[packetIdx])
+	}
+
+	require.Nil(t, d.push(build(0, 0, FrameMarkerStart), opts).frame)
+	require.Nil(t, d.push(build(1, 0, FrameMarkerStart), opts).frame)
+	require.Nil(t, d.push(build(2, 0, FrameMarkerStart), opts).frame)
+	require.Nil(t, d.push(build(0, 1, FrameMarkerInter), opts).frame)
+	require.Nil(t, d.push(build(1, 1, FrameMarkerInter), opts).frame)
+	require.Nil(t, d.push(build(2, 1, FrameMarkerInter), opts).frame)
+
+	frameTwo := d.push(build(1, 2, FrameMarkerFinal), opts).frame
+	require.NotNil(t, frameTwo)
+	require.Equal(t, []byte{0xb1, 0xb2, 0xb3}, frameTwo.payload)
+
+	frameOne := d.push(build(0, 2, FrameMarkerFinal), opts).frame
+	require.NotNil(t, frameOne)
+	require.Equal(t, []byte{0xa1, 0xa2, 0xa3}, frameOne.payload)
+
+	frameThree := d.push(build(2, 2, FrameMarkerFinal), opts).frame
+	require.NotNil(t, frameThree)
+	require.Equal(t, []byte{0xc1, 0xc2, 0xc3}, frameThree.payload)
+}
+
+func TestDepacketizer_MaxPartialFramesChangeAcrossPushes(t *testing.T) {
+	d := newDepacketizer()
+	opts := depacketizerPushOptions{maxPartialFrames: 2}
+
+	base := testPacket()
+	startFor := func(frameNumber uint16) dtp.Packet {
+		return derive(base, FrameMarkerStart, frameNumber*100, frameNumber)
+	}
+	finalFor := func(frameNumber uint16) dtp.Packet {
+		return derive(base, FrameMarkerFinal, frameNumber*100+1, frameNumber)
+	}
+
+	require.Nil(t, d.push(startFor(1), opts).frame)
+	require.Nil(t, d.push(startFor(2), opts).frame)
+
+	// expanding the cap admits frames 3 and 4 without evicting anything
+	opts.maxPartialFrames = 4
+
+	result := d.push(startFor(3), opts)
+	require.Nil(t, result.frame)
+	require.Nil(t, result.drop)
+
+	result = d.push(startFor(4), opts)
+	require.Nil(t, result.frame)
+	require.Nil(t, result.drop)
+
+	require.NotNil(t, d.push(finalFor(1), opts).frame)
+
+	// shrinking the cap evicts frames 2 and 3 when frame 5 starts; only the first eviction is reported
+	opts.maxPartialFrames = 2
+
+	require.Nil(t, d.push(startFor(5), opts).frame)
+
+	result = d.push(finalFor(2), opts)
+	require.Equal(t, &depacketizerDropError{frameNumber: 2, reason: depacketizerDropReasonUnknownFrame}, result.drop)
+
+	result = d.push(finalFor(3), opts)
+	require.Equal(t, &depacketizerDropError{frameNumber: 3, reason: depacketizerDropReasonUnknownFrame}, result.drop)
+
+	require.NotNil(t, d.push(finalFor(4), opts).frame)
+	require.NotNil(t, d.push(finalFor(5), opts).frame)
+}
+
+func TestDepacketizerDropError_Error(t *testing.T) {
+	require.Equal(t, "frame 7 dropped: incomplete (2/4)", fmt.Sprint(&depacketizerDropError{
+		frameNumber: 7, reason: depacketizerDropReasonIncomplete, received: 2, expected: 4,
+	}))
+}
diff --git a/datatrack/frame.go b/datatrack/frame.go
new file mode 100644
index 00000000..1b63f0d7
--- /dev/null
+++ b/datatrack/frame.go
@@ -0,0 +1,45 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package datatrack
+
+import "time"
+
+// Frame is a frame published on a data track.
+type Frame struct {
+	Payload []byte
+	// UserTimestamp is an optional application-defined timestamp; nil when unset.
+	UserTimestamp *uint64
+}
+
+// UserTimestampNow returns the current Unix time in milliseconds, the interpretation
+// DurationSinceTimestamp expects.
+func UserTimestampNow() *uint64 {
+	timestamp := uint64(time.Now().UnixMilli())
+	return &timestamp
+}
+
+// DurationSinceTimestamp returns how much time has passed since the frame's user timestamp,
+// taken as Unix milliseconds. It is false if the frame has no user timestamp or the timestamp
+// lies in the future.
+func (f Frame) DurationSinceTimestamp() (time.Duration, bool) {
+	if f.UserTimestamp == nil || *f.UserTimestamp > uint64(1<<63-1) {
+		return 0, false
+	}
+	elapsed := time.Since(time.UnixMilli(int64(*f.UserTimestamp)))
+	if elapsed < 0 {
+		return 0, false
+	}
+	return elapsed, true
+}
diff --git a/datatrack/handle.go b/datatrack/handle.go
new file mode 100644
index 00000000..566a4009
--- /dev/null
+++ b/datatrack/handle.go
@@ -0,0 +1,52 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package datatrack
+
+import (
+	"errors"
+	"math"
+)
+
+var (
+	errHandleReserved   = errors.New("data track handle 0 is reserved")
+	errHandleTooLarge   = errors.New("value too large to be a data track handle")
+	errHandlesExhausted = errors.New("data track handles exhausted")
+)
+
+// trackHandle identifies the data track a packet belongs to.
+type trackHandle uint16
+
+func handleFromUint32(value uint32) (trackHandle, error) {
+	if value > math.MaxUint16 {
+		return 0, errHandleTooLarge
+	}
+	if value == 0 {
+		return 0, errHandleReserved
+	}
+	return trackHandle(value), nil
+}
+
+// handleAllocator hands out unique handles for local publications. Handles are never reused.
+type handleAllocator struct {
+	value uint16
+}
+
+func (a *handleAllocator) get() (trackHandle, error) {
+	if a.value == math.MaxUint16 {
+		return 0, errHandlesExhausted
+	}
+	a.value++
+	return trackHandle(a.value), nil
+}
diff --git a/datatrack/info.go b/datatrack/info.go
new file mode 100644
index 00000000..64eaf847
--- /dev/null
+++ b/datatrack/info.go
@@ -0,0 +1,43 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package datatrack
+
+import (
+	"errors"
+	"strings"
+
+	"github.com/livekit/protocol/livekit"
+	"github.com/livekit/protocol/utils/guid"
+)
+
+var ErrInvalidSID = errors.New("invalid data track SID")
+
+func parseSID(raw string) (livekit.TrackID, error) {
+	if !strings.HasPrefix(raw, guid.DataTrackPrefix) {
+		return "", ErrInvalidSID
+	}
+	return livekit.TrackID(raw), nil
+}
+
+// Info describes a published data track. The SID changes when the publisher completes a
+// full reconnect; Name is stable. Schema and FrameEncoding are nil when not specified.
+type Info struct {
+	SID           livekit.TrackID
+	pubHandle     trackHandle
+	Name          string
+	UsesE2EE      bool
+	Schema        *SchemaID
+	FrameEncoding FrameEncoding
+}
diff --git a/datatrack/packet.go b/datatrack/packet.go
new file mode 100644
index 00000000..08a1d0de
--- /dev/null
+++ b/datatrack/packet.go
@@ -0,0 +1,117 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package datatrack
+
+import (
+	"errors"
+	"fmt"
+
+	dtp "github.com/livekit/protocol/datatrack"
+)
+
+const supportedVersion = 0
+
+var (
+	ErrUnsupportedVersion = errors.New("unsupported data track packet version")
+	ErrInvalidHandle      = errors.New("invalid data track handle")
+)
+
+// FrameMarker is a packet's position within its frame.
+type FrameMarker uint8
+
+const (
+	FrameMarkerInter FrameMarker = iota
+	FrameMarkerStart
+	FrameMarkerFinal
+	FrameMarkerSingle
+)
+
+func markerOf(h *dtp.Header) FrameMarker {
+	switch {
+	case h.IsStartOfFrame && h.IsFinalOfFrame:
+		return FrameMarkerSingle
+	case h.IsStartOfFrame:
+		return FrameMarkerStart
+	case h.IsFinalOfFrame:
+		return FrameMarkerFinal
+	default:
+		return FrameMarkerInter
+	}
+}
+
+func (m FrameMarker) apply(h *dtp.Header) {
+	h.IsStartOfFrame = m == FrameMarkerStart || m == FrameMarkerSingle
+	h.IsFinalOfFrame = m == FrameMarkerFinal || m == FrameMarkerSingle
+}
+
+// Extensions are the header extensions understood by this SDK.
+type Extensions struct {
+	UserTimestamp *uint64
+	E2EE          *dtp.ExtensionE2EE
+}
+
+func (e Extensions) apply(h *dtp.Header) error {
+	if e.E2EE != nil {
+		ext, err := e.E2EE.Marshal()
+		if err != nil {
+			return err
+		}
+		h.AddExtension(ext)
+	}
+	if e.UserTimestamp != nil {
+		ext, err := dtp.NewExtensionUserTimestamp(*e.UserTimestamp).Marshal()
+		if err != nil {
+			return err
+		}
+		h.AddExtension(ext)
+	}
+	return nil
+}
+
+// extensionsOf reads the known extensions. Unknown ids and known ids that fail to
+// unmarshal (e.g. less than the expected data) are skipped.
+func extensionsOf(h *dtp.Header) Extensions {
+	var extensions Extensions
+	for _, ext := range h.Extensions {
+		switch ext.ID() {
+		case dtp.ExtensionE2EEID:
+			var e2ee dtp.ExtensionE2EE
+			if e2ee.Unmarshal(ext) == nil {
+				extensions.E2EE = &e2ee
+			}
+		case dtp.ExtensionUserTimestampID:
+			var userTimestamp dtp.ExtensionUserTimestamp
+			if userTimestamp.Unmarshal(ext) == nil {
+				timestamp := userTimestamp.Timestamp()
+				extensions.UserTimestamp = &timestamp
+			}
+		}
+	}
+	return extensions
+}
+
+func parsePacket(buf []byte) (*dtp.Packet, error) {
+	var packet dtp.Packet
+	if err := packet.Unmarshal(buf); err != nil {
+		return nil, err
+	}
+	if packet.Version > supportedVersion {
+		return nil, fmt.Errorf("%w: %d", ErrUnsupportedVersion, packet.Version)
+	}
+	if packet.Handle == 0 {
+		return nil, fmt.Errorf("%w: 0 is reserved", ErrInvalidHandle)
+	}
+	return &packet, nil
+}
diff --git a/datatrack/packet_test.go b/datatrack/packet_test.go
new file mode 100644
index 00000000..aa120dcf
--- /dev/null
+++ b/datatrack/packet_test.go
@@ -0,0 +1,205 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package datatrack
+
+import (
+	"bytes"
+	"encoding/binary"
+	"testing"
+	"time"
+
+	dtp "github.com/livekit/protocol/datatrack"
+	"github.com/livekit/protocol/utils/pointer"
+	"github.com/stretchr/testify/require"
+)
+
+// testHeader is the header of the packet used by the Rust serialization tests, without extensions.
+func testHeader() dtp.Header {
+	header := dtp.Header{
+		Version:        supportedVersion,
+		Handle:         0x8811,
+		SequenceNumber: 0x4422,
+		FrameNumber:    0x4411,
+		Timestamp:      0x44221188,
+	}
+	FrameMarkerFinal.apply(&header)
+	return header
+}
+
+func testExtensions() Extensions {
+	var iv [dtp.ExtensionE2EEIVLength]byte
+	for i := range iv {
+		iv[i] = 0x3c
+	}
+	return Extensions{
+		UserTimestamp: pointer.To(uint64(0x4411221111118811)),
+		E2EE:          dtp.NewExtensionE2EE(0xfa, iv),
+	}
+}
+
+// testPacket is the packet used by the Rust serialization tests.
+func testPacket() dtp.Packet {
+	header := testHeader()
+	if err := testExtensions().apply(&header); err != nil {
+		panic(err)
+	}
+	return dtp.Packet{Header: header, Payload: bytes.Repeat([]byte{0xfa}, 1024)}
+}
+
+// validRawPacket is the simplest valid raw packet: a base header with a non-zero handle.
+func validRawPacket() []byte {
+	raw := make([]byte, 12)
+	raw[3] = 1
+	return raw
+}
+
+func withExtensionWords(raw []byte, words uint16) []byte {
+	raw[0] |= 1 << 2
+	return binary.BigEndian.AppendUint16(raw, words)
+}
+
+func TestPacket_Roundtrip(t *testing.T) {
+	extensions := testExtensions()
+	for _, tc := range []struct {
+		name       string
+		marker     FrameMarker
+		extensions Extensions
+	}{
+		{"no_extensions", FrameMarkerSingle, Extensions{}},
+		{"user_timestamp", FrameMarkerStart, Extensions{UserTimestamp: extensions.UserTimestamp}},
+		{"e2ee", FrameMarkerInter, Extensions{E2EE: extensions.E2EE}},
+		{"all_extensions", FrameMarkerFinal, extensions},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			header := testHeader()
+			tc.marker.apply(&header)
+			require.NoError(t, tc.extensions.apply(&header))
+			original := dtp.Packet{Header: header, Payload: bytes.Repeat([]byte{0xab}, 300)}
+
+			raw, err := original.Marshal()
+			require.NoError(t, err)
+			parsed, err := parsePacket(raw)
+			require.NoError(t, err)
+
+			require.Equal(t, original.Header, parsed.Header)
+			require.Equal(t, original.Payload, parsed.Payload)
+			require.Equal(t, tc.extensions, extensionsOf(&parsed.Header))
+		})
+	}
+}
+
+func TestPacket_Serialize(t *testing.T) {
+	packet := testPacket()
+	raw, err := packet.Marshal()
+	require.NoError(t, err)
+	require.Len(t, raw, 1064)
+
+	expectedHeader := []byte{
+		0x0c, 0x00, // version 0, final, extensions; reserved
+		0x88, 0x11, // handle
+		0x44, 0x22, // sequence
+		0x44, 0x11, // frame number
+		0x44, 0x22, 0x11, 0x88, // timestamp
+		0x00, 0x06, // extension words
+		0x01, 0x0d, 0xfa, 0x3c, 0x3c, 0x3c, 0x3c, 0x3c, 0x3c, 0x3c, 0x3c, 0x3c, 0x3c, 0x3c, 0x3c, // E2EE
+		0x02, 0x08, 0x44, 0x11, 0x22, 0x11, 0x11, 0x11, 0x88, 0x11, // user timestamp
+		0x00, // padding
+	}
+	require.Equal(t, expectedHeader, raw[:len(expectedHeader)])
+	require.Equal(t, bytes.Repeat([]byte{0xfa}, 1024), raw[len(expectedHeader):])
+}
+
+func TestPacket_UnsupportedVersion(t *testing.T) {
+	raw := validRawPacket()
+	raw[0] = 0x20
+	_, err := parsePacket(raw)
+	require.ErrorIs(t, err, ErrUnsupportedVersion)
+}
+
+func TestPacket_ReservedHandle(t *testing.T) {
+	raw := validRawPacket()
+	raw[3] = 0
+	_, err := parsePacket(raw)
+	require.ErrorIs(t, err, ErrInvalidHandle)
+}
+
+func TestPacket_ExtE2EE(t *testing.T) {
+	raw := withExtensionWords(validRawPacket(), 4)
+	raw = append(raw, 1, 13, 0xfa)
+	raw = append(raw, bytes.Repeat([]byte{0x3c}, 12)...)
+	raw = append(raw, 0, 0, 0)
+
+	packet, err := parsePacket(raw)
+	require.NoError(t, err)
+	e2ee := extensionsOf(&packet.Header).E2EE
+	require.NotNil(t, e2ee)
+	require.Equal(t, uint8(0xfa), e2ee.KeyIndex())
+	iv := e2ee.IV()
+	require.Equal(t, bytes.Repeat([]byte{0x3c}, 12), iv[:])
+}
+
+func TestPacket_ExtUserTimestamp(t *testing.T) {
+	raw := withExtensionWords(validRawPacket(), 2)
+	raw = append(raw, 2, 8, 0x44, 0x11, 0x22, 0x11, 0x11, 0x11, 0x88, 0x11)
+
+	packet, err := parsePacket(raw)
+	require.NoError(t, err)
+	require.Equal(t, pointer.To(uint64(0x4411221111118811)), extensionsOf(&packet.Header).UserTimestamp)
+}
+
+func TestPacket_ExtForwardCompatLongerLength(t *testing.T) {
+	raw := withExtensionWords(validRawPacket(), 3)
+	raw = append(raw, 2, 12, 0x44, 0x11, 0x22, 0x11, 0x11, 0x11, 0x88, 0x11) // known 8 bytes
+	raw = append(raw, 0xff, 0xff, 0xff, 0xff)                                // extra bytes from a future version
+
+	packet, err := parsePacket(raw)
+	require.NoError(t, err)
+	require.Equal(t, pointer.To(uint64(0x4411221111118811)), extensionsOf(&packet.Header).UserTimestamp)
+}
+
+func TestPacket_ExtShorterThanKnownLengthSkipped(t *testing.T) {
+	raw := withExtensionWords(validRawPacket(), 1)
+	raw = append(raw, 2, 4, 0x3c, 0x3c, 0x3c, 0x3c)
+
+	packet, err := parsePacket(raw)
+	require.NoError(t, err)
+	require.Nil(t, extensionsOf(&packet.Header).UserTimestamp)
+}
+
+func TestClock_IsBaseAtEpoch(t *testing.T) {
+	epoch := time.Now()
+	base := timestamp(1234)
+	c := newClockWithEpoch(epoch, base)
+
+	require.Equal(t, base, c.at(epoch))
+	require.Equal(t, base, c.prev)
+}
+
+func TestTimestamp_IsBeforeHalfRange(t *testing.T) {
+	for _, a := range []timestamp{0, 1, 1 << 31, 0xffffffff} {
+		b := a.wrappingAdd(1 << 31)
+		require.NotEqual(t, a.isBefore(b), b.isBefore(a), "a=%#x b=%#x", a, b)
+		require.Equal(t, a < b, a.isBefore(b), "a=%#x b=%#x", a, b)
+	}
+}
+
+func TestClock_Monotonic(t *testing.T) {
+	epoch := time.Now()
+	c := newClockWithEpoch(epoch, 0)
+
+	t1 := epoch.Add(100 * time.Millisecond)
+	t0 := epoch.Add(50 * time.Millisecond)
+	require.Equal(t, c.at(t1), c.at(t0), "clock went backwards")
+}
diff --git a/datatrack/packetizer.go b/datatrack/packetizer.go
new file mode 100644
index 00000000..505e2128
--- /dev/null
+++ b/datatrack/packetizer.go
@@ -0,0 +1,89 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package datatrack
+
+import (
+	"errors"
+
+	dtp "github.com/livekit/protocol/datatrack"
+)
+
+var errMTUTooShort = errors.New("MTU is too short to send frame")
+
+// packetizer converts frames into packets for transport.
+type packetizer struct {
+	handle      trackHandle
+	mtu         int
+	sequence    uint16
+	frameNumber uint16
+	clock       *clock
+}
+
+func newPacketizer(handle trackHandle, mtu int) *packetizer {
+	return &packetizer{handle: handle, mtu: mtu, clock: newClock(randomTimestamp())}
+}
+
+func (p *packetizer) packetize(payload []byte, extensions Extensions) ([]dtp.Packet, error) {
+	header := dtp.Header{
+		Version:   supportedVersion,
+		Handle:    uint16(p.handle),
+		Timestamp: uint32(p.clock.now()),
+	}
+	if err := extensions.apply(&header); err != nil {
+		return nil, err
+	}
+
+	maxPayloadSize := p.mtu - header.MarshalSize()
+	if maxPayloadSize <= 0 {
+		return nil, errMTUTooShort
+	}
+	header.FrameNumber = p.frameNumber
+	p.frameNumber++
+
+	chunks := chunkPayload(payload, maxPayloadSize)
+	packets := make([]dtp.Packet, len(chunks))
+	for i, chunk := range chunks {
+		packetHeader := header
+		frameMarker(i, len(chunks)).apply(&packetHeader)
+		packetHeader.SequenceNumber = p.sequence
+		p.sequence++
+		packets[i] = dtp.Packet{Header: packetHeader, Payload: chunk}
+	}
+	return packets, nil
+}
+
+func frameMarker(index, packetCount int) FrameMarker {
+	switch {
+	case packetCount <= 1:
+		return FrameMarkerSingle
+	case index == 0:
+		return FrameMarkerStart
+	case index == packetCount-1:
+		return FrameMarkerFinal
+	default:
+		return FrameMarkerInter
+	}
+}
+
+// chunkPayload splits payload into consecutive chunks of at most maxSize bytes without copying.
+func chunkPayload(payload []byte, maxSize int) [][]byte {
+	var chunks [][]byte
+	for len(payload) > 0 {
+		n := min(maxSize, len(payload))
+		chunks = append(chunks, payload[:n])
+		payload = payload[n:]
+	}
+	return chunks
+}
diff --git a/datatrack/packetizer_test.go b/datatrack/packetizer_test.go
new file mode 100644
index 00000000..2b3481ad
--- /dev/null
+++ b/datatrack/packetizer_test.go
@@ -0,0 +1,98 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package datatrack
+
+import (
+	"bytes"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestPacketizer_FrameMarker(t *testing.T) {
+	for _, tc := range []struct {
+		index, packetCount int
+		expected           FrameMarker
+	}{
+		{0, 1, FrameMarkerSingle},
+		{0, 10, FrameMarkerStart},
+		{4, 10, FrameMarkerInter},
+		{9, 10, FrameMarkerFinal},
+	} {
+		t.Run(fmt.Sprintf("index=%d,count=%d", tc.index, tc.packetCount), func(t *testing.T) {
+			require.Equal(t, tc.expected, frameMarker(tc.index, tc.packetCount))
+		})
+	}
+}
+
+func TestPacketizer_Packetize(t *testing.T) {
+	for _, tc := range []struct {
+		name         string
+		payloadSize  int
+		mtuSize      int
+		packetsCount int
+	}{
+		{"zero_payload", 0, 1_024, 0},
+		{"single_packet", 128, 1_024, 1},
+		{"multi_packet", 20_480, 1_024, 21},
+		{"multi_packet_mtu_16000", 40_960, 16_000, 3},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			handle := trackHandle(0x8811)
+			extensions := testExtensions()
+
+			p := newPacketizer(handle, tc.mtuSize)
+			packets, err := p.packetize(bytes.Repeat([]byte{0xab}, tc.payloadSize), extensions)
+			require.NoError(t, err)
+
+			if len(packets) == 0 {
+				require.Zero(t, tc.payloadSize, "should be no packets for zero payload")
+				return
+			}
+			for i, packet := range packets {
+				require.Equal(t, frameMarker(i, len(packets)), markerOf(&packet.Header))
+				require.Equal(t, uint16(0), packet.FrameNumber)
+				require.Equal(t, uint16(handle), packet.Handle)
+				require.Equal(t, uint16(i), packet.SequenceNumber)
+				require.Equal(t, extensions, extensionsOf(&packet.Header))
+			}
+		})
+	}
+}
+
+func TestChunkPayload_EmptySource(t *testing.T) {
+	require.Empty(t, chunkPayload(nil, 256))
+}
+
+func TestChunkPayload_Chunks(t *testing.T) {
+	for _, chunkSize := range []int{1, 128, 333} {
+		for _, sourceSize := range []int{1, 64, 128, 256, 123} {
+			t.Run(fmt.Sprintf("chunk=%d,source=%d", chunkSize, sourceSize), func(t *testing.T) {
+				chunks := chunkPayload(bytes.Repeat([]byte{0xcc}, sourceSize), chunkSize)
+
+				require.Len(t, chunks, (sourceSize+chunkSize-1)/chunkSize)
+				for _, chunk := range chunks[:len(chunks)-1] {
+					require.Len(t, chunk, chunkSize)
+				}
+				expectedLastLen := sourceSize % chunkSize
+				if expectedLastLen == 0 {
+					expectedLastLen = min(chunkSize, sourceSize)
+				}
+				require.Len(t, chunks[len(chunks)-1], expectedLastLen)
+			})
+		}
+	}
+}
diff --git a/datatrack/proto.go b/datatrack/proto.go
new file mode 100644
index 00000000..3b438311
--- /dev/null
+++ b/datatrack/proto.go
@@ -0,0 +1,205 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package datatrack
+
+import (
+	"fmt"
+
+	"github.com/livekit/protocol/livekit"
+)
+
+var (
+	wellKnownSchemaEncodingToProto = map[WellKnownSchemaEncoding]livekit.DataTrackSchemaEncoding_WellKnownSchemaEncoding{
+		SchemaEncodingProtobuf:   livekit.DataTrackSchemaEncoding_WELL_KNOWN_SCHEMA_ENCODING_PROTOBUF,
+		SchemaEncodingFlatbuffer: livekit.DataTrackSchemaEncoding_WELL_KNOWN_SCHEMA_ENCODING_FLATBUFFER,
+		SchemaEncodingROS1Msg:    livekit.DataTrackSchemaEncoding_WELL_KNOWN_SCHEMA_ENCODING_ROS1_MSG,
+		SchemaEncodingROS2Msg:    livekit.DataTrackSchemaEncoding_WELL_KNOWN_SCHEMA_ENCODING_ROS2_MSG,
+		SchemaEncodingROS2IDL:    livekit.DataTrackSchemaEncoding_WELL_KNOWN_SCHEMA_ENCODING_ROS2_IDL,
+		SchemaEncodingOMGIDL:     livekit.DataTrackSchemaEncoding_WELL_KNOWN_SCHEMA_ENCODING_OMG_IDL,
+		SchemaEncodingJSONSchema: livekit.DataTrackSchemaEncoding_WELL_KNOWN_SCHEMA_ENCODING_JSON_SCHEMA,
+	}
+	wellKnownSchemaEncodingFromProto = invert(wellKnownSchemaEncodingToProto)
+
+	wellKnownFrameEncodingToProto = map[WellKnownFrameEncoding]livekit.DataTrackFrameEncoding_WellKnownFrameEncoding{
+		FrameEncodingROS1:       livekit.DataTrackFrameEncoding_WELL_KNOWN_FRAME_ENCODING_ROS1,
+		FrameEncodingCDR:        livekit.DataTrackFrameEncoding_WELL_KNOWN_FRAME_ENCODING_CDR,
+		FrameEncodingProtobuf:   livekit.DataTrackFrameEncoding_WELL_KNOWN_FRAME_ENCODING_PROTOBUF,
+		FrameEncodingFlatbuffer: livekit.DataTrackFrameEncoding_WELL_KNOWN_FRAME_ENCODING_FLATBUFFER,
+		FrameEncodingCBOR:       livekit.DataTrackFrameEncoding_WELL_KNOWN_FRAME_ENCODING_CBOR,
+		FrameEncodingMsgPack:    livekit.DataTrackFrameEncoding_WELL_KNOWN_FRAME_ENCODING_MSGPACK,
+		FrameEncodingJSON:       livekit.DataTrackFrameEncoding_WELL_KNOWN_FRAME_ENCODING_JSON,
+	}
+	wellKnownFrameEncodingFromProto = invert(wellKnownFrameEncodingToProto)
+)
+
+func invert[K, V comparable](m map[K]V) map[V]K {
+	inverted := make(map[V]K, len(m))
+	for k, v := range m {
+		inverted[v] = k
+	}
+	return inverted
+}
+
+// schemaEncodingFromProto maps unspecified and unknown well-known values to SchemaEncodingOther.
+func schemaEncodingFromProto(msg *livekit.DataTrackSchemaEncoding) SchemaEncoding {
+	switch value := msg.GetValue().(type) {
+	case *livekit.DataTrackSchemaEncoding_WellKnown:
+		if encoding, known := wellKnownSchemaEncodingFromProto[value.WellKnown]; known {
+			return encoding
+		}
+		return SchemaEncodingOther
+	case *livekit.DataTrackSchemaEncoding_Custom:
+		return CustomSchemaEncoding(value.Custom)
+	default:
+		return SchemaEncodingOther
+	}
+}
+
+// schemaEncodingToProto maps SchemaEncodingOther and nil to the unspecified well-known value.
+func schemaEncodingToProto(encoding SchemaEncoding) *livekit.DataTrackSchemaEncoding {
+	if custom, ok := encoding.(CustomSchemaEncoding); ok {
+		return &livekit.DataTrackSchemaEncoding{Value: &livekit.DataTrackSchemaEncoding_Custom{Custom: string(custom)}}
+	}
+	wellKnown, _ := encoding.(WellKnownSchemaEncoding)
+	return &livekit.DataTrackSchemaEncoding{Value: &livekit.DataTrackSchemaEncoding_WellKnown{
+		WellKnown: wellKnownSchemaEncodingToProto[wellKnown],
+	}}
+}
+
+func schemaIDFromProto(msg *livekit.DataTrackSchemaId) SchemaID {
+	id := SchemaID{Name: msg.GetName(), Encoding: SchemaEncodingOther}
+	if msg.GetEncoding() != nil {
+		id.Encoding = schemaEncodingFromProto(msg.GetEncoding())
+	}
+	return id
+}
+
+func schemaIDToProto(id SchemaID) *livekit.DataTrackSchemaId {
+	return &livekit.DataTrackSchemaId{Name: id.Name, Encoding: schemaEncodingToProto(id.Encoding)}
+}
+
+// frameEncodingFromProto maps unspecified and unknown well-known values to FrameEncodingOther.
+func frameEncodingFromProto(msg *livekit.DataTrackFrameEncoding) FrameEncoding {
+	switch value := msg.GetValue().(type) {
+	case *livekit.DataTrackFrameEncoding_WellKnown:
+		if encoding, known := wellKnownFrameEncodingFromProto[value.WellKnown]; known {
+			return encoding
+		}
+		return FrameEncodingOther
+	case *livekit.DataTrackFrameEncoding_Custom:
+		return CustomFrameEncoding(value.Custom)
+	default:
+		return FrameEncodingOther
+	}
+}
+
+// frameEncodingToProto maps FrameEncodingOther and nil to the unspecified well-known value.
+func frameEncodingToProto(encoding FrameEncoding) *livekit.DataTrackFrameEncoding {
+	if custom, ok := encoding.(CustomFrameEncoding); ok {
+		return &livekit.DataTrackFrameEncoding{Value: &livekit.DataTrackFrameEncoding_Custom{Custom: string(custom)}}
+	}
+	wellKnown, _ := encoding.(WellKnownFrameEncoding)
+	return &livekit.DataTrackFrameEncoding{Value: &livekit.DataTrackFrameEncoding_WellKnown{
+		WellKnown: wellKnownFrameEncodingToProto[wellKnown],
+	}}
+}
+
+func infoFromProto(msg *livekit.DataTrackInfo) (Info, error) {
+	handle, err := handleFromUint32(msg.GetPubHandle())
+	if err != nil {
+		return Info{}, err
+	}
+
+	var usesE2EE bool
+	switch msg.GetEncryption() {
+	case livekit.Encryption_NONE:
+	case livekit.Encryption_GCM:
+		usesE2EE = true
+	default:
+		return Info{}, fmt.Errorf("unsupported E2EE type: %s", msg.GetEncryption())
+	}
+
+	sid, err := parseSID(msg.GetSid())
+	if err != nil {
+		return Info{}, err
+	}
+
+	info := Info{SID: sid, pubHandle: handle, Name: msg.GetName(), UsesE2EE: usesE2EE}
+	if msg.GetSchema() != nil {
+		schema := schemaIDFromProto(msg.GetSchema())
+		info.Schema = &schema
+	}
+	if msg.GetFrameEncoding() != nil {
+		info.FrameEncoding = frameEncodingFromProto(msg.GetFrameEncoding())
+	}
+	return info, nil
+}
+
+func infoToProto(info Info) *livekit.DataTrackInfo {
+	msg := &livekit.DataTrackInfo{
+		PubHandle:  uint32(info.pubHandle),
+		Sid:        string(info.SID),
+		Name:       info.Name,
+		Encryption: livekit.Encryption_NONE,
+	}
+	if info.UsesE2EE {
+		msg.Encryption = livekit.Encryption_GCM
+	}
+	if info.Schema != nil {
+		msg.Schema = schemaIDToProto(*info.Schema)
+	}
+	if info.FrameEncoding != nil {
+		msg.FrameEncoding = frameEncodingToProto(info.FrameEncoding)
+	}
+	return msg
+}
+
+// publishResponsesForSyncState describes the published tracks for SyncState.
+func publishResponsesForSyncState(published []Info) []*livekit.PublishDataTrackResponse {
+	responses := make([]*livekit.PublishDataTrackResponse, 0, len(published))
+	for _, info := range published {
+		responses = append(responses, &livekit.PublishDataTrackResponse{Info: infoToProto(info)})
+	}
+	return responses
+}
+
+// subscriberHandlesFromProto maps the handles of incoming packets to the tracks they belong to.
+func subscriberHandlesFromProto(msg *livekit.DataTrackSubscriberHandles) (map[trackHandle]livekit.TrackID, error) {
+	mapping := make(map[trackHandle]livekit.TrackID, len(msg.GetSubHandles()))
+	for rawHandle, track := range msg.GetSubHandles() {
+		handle, err := handleFromUint32(rawHandle)
+		if err != nil {
+			return nil, err
+		}
+		sid, err := parseSID(track.GetTrackSid())
+		if err != nil {
+			return nil, err
+		}
+		mapping[handle] = sid
+	}
+	return mapping, nil
+}
+
+// extractTrackInfo returns the data tracks published by a participant.
+func extractTrackInfo(participant *livekit.ParticipantInfo) ([]Info, error) {
+	infos := make([]Info, 0, len(participant.GetDataTracks()))
+	for _, msg := range participant.GetDataTracks() {
+		info, err := infoFromProto(msg)
+		if err != nil {
+			return nil, err
+		}
+		infos = append(infos, info)
+	}
+	return infos, nil
+}
diff --git a/datatrack/proto_test.go b/datatrack/proto_test.go
new file mode 100644
index 00000000..f1e40cf6
--- /dev/null
+++ b/datatrack/proto_test.go
@@ -0,0 +1,117 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package datatrack
+
+import (
+	"testing"
+
+	"github.com/livekit/protocol/livekit"
+	"github.com/stretchr/testify/require"
+)
+
+func testInfo() Info {
+	return Info{SID: "DTR_1234", pubHandle: 1, Name: "track"}
+}
+
+func TestProto_InfoFromProto(t *testing.T) {
+	response := &livekit.PublishDataTrackResponse{
+		Info: &livekit.DataTrackInfo{
+			PubHandle:     1,
+			Sid:           "DTR_1234",
+			Name:          "track",
+			Encryption:    livekit.Encryption_GCM,
+			Schema:        schemaIDToProto(SchemaID{Name: "schema", Encoding: SchemaEncodingJSONSchema}),
+			FrameEncoding: frameEncodingToProto(FrameEncodingJSON),
+		},
+	}
+
+	info, err := infoFromProto(response.Info)
+	require.NoError(t, err)
+	require.Equal(t, trackHandle(1), info.pubHandle)
+	require.Equal(t, livekit.TrackID("DTR_1234"), info.SID)
+	require.Equal(t, "track", info.Name)
+	require.Equal(t, &SchemaID{Name: "schema", Encoding: SchemaEncodingJSONSchema}, info.Schema)
+	require.Equal(t, FrameEncodingJSON, info.FrameEncoding)
+	require.True(t, info.UsesE2EE)
+}
+
+func TestProto_FrameEncodingMapping(t *testing.T) {
+	base := &livekit.DataTrackInfo{
+		PubHandle:  1,
+		Sid:        "DTR_1234",
+		Name:       "track",
+		Encryption: livekit.Encryption_NONE,
+	}
+
+	info, err := infoFromProto(base)
+	require.NoError(t, err)
+	require.Nil(t, info.FrameEncoding)
+
+	unspecified := &livekit.DataTrackInfo{FrameEncoding: frameEncodingToProto(FrameEncodingOther)}
+	unspecified.PubHandle, unspecified.Sid, unspecified.Name = base.PubHandle, base.Sid, base.Name
+	info, err = infoFromProto(unspecified)
+	require.NoError(t, err)
+	require.Equal(t, FrameEncodingOther, info.FrameEncoding)
+
+	custom := &livekit.DataTrackInfo{FrameEncoding: frameEncodingToProto(CustomFrameEncoding("my_encoding"))}
+	custom.PubHandle, custom.Sid, custom.Name = base.PubHandle, base.Sid, base.Name
+	info, err = infoFromProto(custom)
+	require.NoError(t, err)
+	require.Equal(t, CustomFrameEncoding("my_encoding"), info.FrameEncoding)
+}
+
+func TestProto_PublishResponsesForSyncState(t *testing.T) {
+	first := testInfo()
+	first.UsesE2EE = true
+
+	second := testInfo()
+	second.UsesE2EE = false
+
+	responses := publishResponsesForSyncState([]Info{first, second})
+	require.Equal(t, livekit.Encryption_GCM, responses[0].Info.Encryption)
+	require.Equal(t, livekit.Encryption_NONE, responses[1].Info.Encryption)
+}
+
+func TestProto_SubscriberHandlesFromProto(t *testing.T) {
+	subscriberHandles := &livekit.DataTrackSubscriberHandles{
+		SubHandles: map[uint32]*livekit.DataTrackSubscriberHandles_PublishedDataTrack{
+			1: {TrackSid: "DTR_1234"},
+			2: {TrackSid: "DTR_4567"},
+		},
+	}
+
+	mapping, err := subscriberHandlesFromProto(subscriberHandles)
+	require.NoError(t, err)
+	require.Equal(t, livekit.TrackID("DTR_1234"), mapping[1])
+	require.Equal(t, livekit.TrackID("DTR_4567"), mapping[2])
+}
+
+func TestProto_ExtractTrackInfo(t *testing.T) {
+	participant := &livekit.ParticipantInfo{
+		DataTracks: []*livekit.DataTrackInfo{{
+			PubHandle:  1,
+			Sid:        "DTR_1234",
+			Name:       "track1",
+			Encryption: livekit.Encryption_GCM,
+		}},
+	}
+
+	infos, err := extractTrackInfo(participant)
+	require.NoError(t, err)
+	require.Len(t, infos, 1)
+	require.Equal(t, trackHandle(1), infos[0].pubHandle)
+	require.Equal(t, "track1", infos[0].Name)
+	require.Equal(t, livekit.TrackID("DTR_1234"), infos[0].SID)
+}
diff --git a/datatrack/schema.go b/datatrack/schema.go
new file mode 100644
index 00000000..25eaab37
--- /dev/null
+++ b/datatrack/schema.go
@@ -0,0 +1,168 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package datatrack
+
+import (
+	"errors"
+	"fmt"
+)
+
+var (
+	ErrSchemaMissingFrameEncoding = errors.New("frame encoding is required when providing schema ID")
+	ErrSchemaMissingID            = errors.New("schema ID is required for frame encoding that is not self-describing")
+	ErrSchemaIncompatible         = errors.New("specified schema and frame encodings are incompatible")
+)
+
+// SchemaID identifies a data track schema. IDs with the same name but different encodings are distinct.
+type SchemaID struct {
+	Name     string
+	Encoding SchemaEncoding
+}
+
+// SchemaEncoding is the encoding of a schema definition: a WellKnownSchemaEncoding or a
+// CustomSchemaEncoding.
+type SchemaEncoding interface {
+	fmt.Stringer
+	isSchemaEncoding()
+}
+
+// WellKnownSchemaEncoding is a schema encoding defined by the protocol.
+type WellKnownSchemaEncoding uint8
+
+const (
+	// SchemaEncodingOther is a well-known encoding not known to this SDK version.
+	SchemaEncodingOther WellKnownSchemaEncoding = iota
+	SchemaEncodingProtobuf
+	SchemaEncodingFlatbuffer
+	SchemaEncodingROS1Msg
+	SchemaEncodingROS2Msg
+	SchemaEncodingROS2IDL
+	SchemaEncodingOMGIDL
+	SchemaEncodingJSONSchema
+)
+
+var wellKnownSchemaEncodingNames = [...]string{"other", "protobuf", "flatbuffer", "ros1msg", "ros2msg", "ros2idl", "omgidl", "jsonschema"}
+
+func (WellKnownSchemaEncoding) isSchemaEncoding() {}
+
+func (e WellKnownSchemaEncoding) String() string {
+	if int(e) < len(wellKnownSchemaEncodingNames) {
+		return wellKnownSchemaEncodingNames[e]
+	}
+	return fmt.Sprintf("WellKnownSchemaEncoding(%d)", uint8(e))
+}
+
+// CustomSchemaEncoding is an application-defined schema encoding identified by name.
+type CustomSchemaEncoding string
+
+func (CustomSchemaEncoding) isSchemaEncoding() {}
+
+func (e CustomSchemaEncoding) String() string {
+	return string(e)
+}
+
+// FrameEncoding is the encoding of frames pushed on a data track: a WellKnownFrameEncoding or a
+// CustomFrameEncoding.
+type FrameEncoding interface {
+	fmt.Stringer
+	// isSelfDescribing reports whether frames need no schema; ok is false when this cannot be determined.
+	isSelfDescribing() (selfDescribing, ok bool)
+	// isDescribedBy reports whether a schema with the given encoding can describe these frames;
+	// ok is false when this cannot be determined.
+	isDescribedBy(schema SchemaEncoding) (described, ok bool)
+}
+
+// WellKnownFrameEncoding is a frame encoding defined by the protocol.
+type WellKnownFrameEncoding uint8
+
+const (
+	// FrameEncodingOther is a well-known encoding not known to this SDK version.
+	FrameEncodingOther WellKnownFrameEncoding = iota
+	FrameEncodingROS1
+	FrameEncodingCDR
+	FrameEncodingProtobuf
+	FrameEncodingFlatbuffer
+	FrameEncodingCBOR
+	FrameEncodingMsgPack
+	FrameEncodingJSON
+)
+
+var wellKnownFrameEncodingNames = [...]string{"other", "ros1", "cdr", "protobuf", "flatbuffer", "cbor", "msgpack", "json"}
+
+func (e WellKnownFrameEncoding) String() string {
+	if int(e) < len(wellKnownFrameEncodingNames) {
+		return wellKnownFrameEncodingNames[e]
+	}
+	return fmt.Sprintf("WellKnownFrameEncoding(%d)", uint8(e))
+}
+
+func (e WellKnownFrameEncoding) isSelfDescribing() (selfDescribing, ok bool) {
+	switch e {
+	case FrameEncodingCBOR, FrameEncodingMsgPack, FrameEncodingJSON:
+		return true, true
+	case FrameEncodingOther:
+		return false, false
+	default:
+		return false, true
+	}
+}
+
+func (e WellKnownFrameEncoding) isDescribedBy(schema SchemaEncoding) (described, ok bool) {
+	if e == FrameEncodingOther {
+		return false, false
+	}
+	switch {
+	case e == FrameEncodingROS1 && schema == SchemaEncodingROS1Msg,
+		e == FrameEncodingCDR && schema == SchemaEncodingROS2Msg,
+		e == FrameEncodingCDR && schema == SchemaEncodingROS2IDL,
+		e == FrameEncodingCDR && schema == SchemaEncodingOMGIDL,
+		e == FrameEncodingProtobuf && schema == SchemaEncodingProtobuf,
+		e == FrameEncodingFlatbuffer && schema == SchemaEncodingFlatbuffer,
+		e == FrameEncodingJSON && schema == SchemaEncodingJSONSchema:
+		return true, true
+	}
+	return false, true
+}
+
+// CustomFrameEncoding is an application-defined frame encoding identified by name.
+type CustomFrameEncoding string
+
+func (e CustomFrameEncoding) String() string {
+	return string(e)
+}
+
+func (CustomFrameEncoding) isSelfDescribing() (selfDescribing, ok bool) {
+	return false, false
+}
+
+func (CustomFrameEncoding) isDescribedBy(SchemaEncoding) (described, ok bool) {
+	return false, false
+}
+
+// validateSchema checks that the given frame and schema encodings are compatible; nil means unset.
+func validateSchema(frameEncoding FrameEncoding, schemaEncoding SchemaEncoding) error {
+	switch {
+	case frameEncoding == nil && schemaEncoding != nil:
+		return ErrSchemaMissingFrameEncoding
+	case frameEncoding != nil && schemaEncoding == nil:
+		if selfDescribing, ok := frameEncoding.isSelfDescribing(); ok && !selfDescribing {
+			return ErrSchemaMissingID
+		}
+	case frameEncoding != nil && schemaEncoding != nil:
+		if described, ok := frameEncoding.isDescribedBy(schemaEncoding); ok && !described {
+			return ErrSchemaIncompatible
+		}
+	}
+	return nil
+}
diff --git a/datatrack/schema_test.go b/datatrack/schema_test.go
new file mode 100644
index 00000000..d3127a28
--- /dev/null
+++ b/datatrack/schema_test.go
@@ -0,0 +1,52 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package datatrack
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestValidateSchema_NotSpecified(t *testing.T) {
+	require.NoError(t, validateSchema(nil, nil))
+}
+
+func TestValidateSchema_SelfDescribing(t *testing.T) {
+	require.NoError(t, validateSchema(FrameEncodingJSON, nil))
+}
+
+func TestValidateSchema_CompatibleEncodings(t *testing.T) {
+	require.NoError(t, validateSchema(FrameEncodingCDR, SchemaEncodingROS2IDL))
+}
+
+func TestValidateSchema_Custom(t *testing.T) {
+	require.NoError(t, validateSchema(
+		CustomFrameEncoding("my-frame-encoding"),
+		CustomSchemaEncoding("my-schema-encoding"),
+	))
+}
+
+func TestValidateSchema_MissingFrameEncoding(t *testing.T) {
+	require.ErrorIs(t, validateSchema(nil, SchemaEncodingProtobuf), ErrSchemaMissingFrameEncoding)
+}
+
+func TestValidateSchema_MissingSchemaID(t *testing.T) {
+	require.ErrorIs(t, validateSchema(FrameEncodingProtobuf, nil), ErrSchemaMissingID)
+}
+
+func TestValidateSchema_Incompatible(t *testing.T) {
+	require.ErrorIs(t, validateSchema(FrameEncodingJSON, SchemaEncodingProtobuf), ErrSchemaIncompatible)
+}
diff --git a/go.mod b/go.mod
index d8c02f29..ec8bc220 100644
--- a/go.mod
+++ b/go.mod
@@ -10,7 +10,7 @@ require (
 	github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731
 	github.com/livekit/media-sdk v0.1.1
 	github.com/livekit/mediatransportutil v0.0.0-20261014192331-dd8e31a0adc0
-	github.com/livekit/protocol v1.52.1
+	github.com/livekit/protocol v1.52.2-0.20261006000721-750fff5dfc0b
 	github.com/magefile/mage v1.17.2
 	github.com/moby/buildkit v0.33.1
 	github.com/moby/patternmatcher v0.6.1
diff --git a/go.sum b/go.sum
index f500fea7..3e3fe4b1 100644
--- a/go.sum
+++ b/go.sum
@@ -139,8 +139,8 @@ github.com/livekit/media-sdk v0.1.1 h1:d0glitmWU5ncIgsW/6wG4Q4iOEcpwINGl85mQjwnK
 github.com/livekit/media-sdk v0.1.1/go.mod h1:TuYRjSepaakL6ATsM9V2VMuksewW1PlhA32BG7Pxty0=
 github.com/livekit/mediatransportutil v0.0.0-20261014192331-dd8e31a0adc0 h1:BSLyHr3Lsl71bHGHsfiYNmbohugR2yoaf9HjAZeFJnc=
 github.com/livekit/mediatransportutil v0.0.0-20261014192331-dd8e31a0adc0/go.mod h1:o8CFmAdrVwzJNOCsQCLUzXRjokkufNshnQHOe4fRaqU=
-github.com/livekit/protocol v1.52.1 h1:w8wRjRlj/57AJBxwomEzpEMzf+lT0qOoLQLO7awdTMU=
-github.com/livekit/protocol v1.52.1/go.mod h1:t5CV/LCsWbI9zsSTHTJKDyLtByqvvVkioSP7jgqpIGA=
+github.com/livekit/protocol v1.52.2-0.20261006000721-750fff5dfc0b h1:8F84iX/JTM5CG/5oDojrAudCkc6RBcjgUl7X407o2vc=
+github.com/livekit/protocol v1.52.2-0.20261006000721-750fff5dfc0b/go.mod h1:t5CV/LCsWbI9zsSTHTJKDyLtByqvvVkioSP7jgqpIGA=
 github.com/livekit/psrpc v0.8.0 h1:uo3x+DL8pK2OIaCxg/s9sLUO2cIocS5HwdPhhI+m+fQ=
 github.com/livekit/psrpc v0.8.0/go.mod h1:6KnU8vULoizX0Er9NNfeKJDV8ug87j3fuo19M0bhj6Q=
 github.com/mackerelio/go-osstat v0.2.8 h1:I2duicTaCGWoM53XwAwA9OIe1inu0xnVs8/pqOWWVr4=