diff --git a/runtime/envoy/session/relay_linux.go b/runtime/envoy/session/relay_linux.go
index a15d54b7..0fb4b97a 100644
--- a/runtime/envoy/session/relay_linux.go
+++ b/runtime/envoy/session/relay_linux.go
@@ -91,6 +91,7 @@ type Relay struct {
 	phase                          context.Context
 	writerCancel                   context.CancelCauseFunc
 	reader                         *relayReader
+	split                          *splitRelay
 	wake, done, readEnded, closing chan struct{}
 	barrier, inputWriter           chan struct{}
 	err, readErr                   error
@@ -130,6 +131,10 @@ func (r *Relay) Err() error                 { r.mu.Lock(); defer r.mu.Unlock();
 func (r *Relay) ReadError() error           { r.mu.Lock(); defer r.mu.Unlock(); return r.readErr }
 
 func (r *Relay) writeOutput(ctx context.Context, data []byte) (err error) {
+	return r.writeOutputStream(ctx, data, "pty")
+}
+
+func (r *Relay) writeOutputStream(ctx context.Context, data []byte, stream string) (err error) {
 	if len(data) == 0 {
 		return nil
 	}
@@ -168,10 +173,16 @@ func (r *Relay) writeOutput(ctx context.Context, data []byte) (err error) {
 			err = errors.Join(err, ctx.Err(), errWorkInterrupted)
 		}
 		if r.closeInterrupted(err) || (err != nil && errors.Is(context.Cause(ctx), errWorkInterrupted)) {
-			r.w.pendingPTY = append(r.w.pendingPTY, original[r.w.OutputThrough-before:]...)
+			pending := &r.w.pendingPTY
+			for i, name := range splitStreams {
+				if stream == name {
+					pending = &r.w.pendingSplit[i]
+				}
+			}
+			*pending = append(*pending, original[r.w.OutputThrough-before:]...)
 		}
 	}()
-	if err := r.w.mark(ctx, false); err != nil {
+	if err := r.w.markStream(ctx, stream, false); err != nil {
 		return err
 	}
 	r.phaseMu.Lock()
@@ -254,6 +265,16 @@ func (r *Relay) run() {
 	for {
 		r.mu.Lock()
 		p := r.reader
+		s := r.split
+		var stdout, stderr <-chan struct{}
+		if s != nil {
+			if s.pumps[0] != nil {
+				stdout = s.pumps[0].wake
+			}
+			if s.pumps[1] != nil {
+				stderr = s.pumps[1].wake
+			}
+		}
 		r.mu.Unlock()
 		var output <-chan struct{}
 		if p != nil {
@@ -278,9 +299,22 @@ func (r *Relay) run() {
 				r.readOutcome(errors.Join(r.err, r.w.channels.Err(), r.reader.err))
 				r.reader = nil
 			}
+			if e := r.joinSplit(); e != nil {
+				r.err = errors.Join(r.err, e)
+				r.w.channels.fail(e)
+			}
 			r.mu.Unlock()
 			return
 		case <-r.wake:
+			r.mu.Lock()
+			if r.ctx.Err() == nil {
+				r.flushSplitPending(r.w.channels.ctx)
+			}
+			r.mu.Unlock()
+		case <-stdout:
+			r.consumeSplit(s, 0)
+		case <-stderr:
+			r.consumeSplit(s, 1)
 		case <-output:
 			r.mu.Lock()
 			if r.ctx.Err() != nil {
diff --git a/runtime/envoy/session/split_relay_linux.go b/runtime/envoy/session/split_relay_linux.go
new file mode 100644
index 00000000..9c1b0d26
--- /dev/null
+++ b/runtime/envoy/session/split_relay_linux.go
@@ -0,0 +1,227 @@
+//go:build linux
+
+package session
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"sync"
+	"syscall"
+	"time"
+)
+
+var splitStreams = [2]string{"stdout", "stderr"}
+
+type splitRelay struct {
+	id     string
+	files  [2]*os.File
+	pumps  [2]*relayReader
+	ended  [2]bool
+	done   chan struct{} // Independent read EOF facts, not output/result completion.
+	once   sync.Once
+	stopIO func()
+}
+
+// The setup owner retains its raw handles. Each poller owns one atomic
+// close-on-exec duplicate of the validated reader, without reopening a path.
+func (s *SplitResources) relayReaders(id string) (files [2]*os.File, err error) {
+	if s == nil || s.operation == nil || s.operation.id != id || len(s.operation.fifos) != 2 {
+		return files, fmt.Errorf("split reader operation mismatch")
+	}
+	if err = s.validate(); err != nil {
+		return files, err
+	}
+	defer func() {
+		if err != nil {
+			for _, file := range files {
+				if file != nil {
+					err = errors.Join(err, file.Close())
+				}
+			}
+		}
+	}()
+	for i, fd := range s.operation.readers {
+		copyFD, _, errno := syscall.Syscall(syscall.SYS_FCNTL, uintptr(fd), syscall.F_DUPFD_CLOEXEC, 0)
+		if errno != 0 {
+			return files, errno
+		}
+		files[i] = os.NewFile(copyFD, splitStreams[i])
+		var st syscall.Stat_t
+		if err = syscall.Fstat(int(copyFD), &st); err != nil {
+			return files, err
+		}
+		if !splitSame(st, s.operation.fifos[i].stat) {
+			return files, fmt.Errorf("split pump descriptor identity changed")
+		}
+	}
+	return files, s.validate()
+}
+
+// attachSplit supplies the transport primitive. B5.3 owns its assembly into
+// start, resize and cleanup; this method does not publish an operation result.
+func (r *Relay) attachSplit(ctx context.Context, owner *SplitResources, id string) (err error) {
+	if err = ctx.Err(); err != nil {
+		return err
+	}
+	files, err := owner.relayReaders(id)
+	if err != nil {
+		return err
+	}
+	attached := false
+	defer func() {
+		if !attached {
+			for _, file := range files {
+				err = errors.Join(err, file.Close())
+			}
+		}
+	}()
+	_, err = r.boundary(ctx, false, func(int64) error {
+		if r.split != nil {
+			return fmt.Errorf("split relay already attached")
+		}
+		s := &splitRelay{id: id, files: files, done: make(chan struct{})}
+		r.split, attached = s, true
+		s.stopIO = joinedAfterFunc(r.ctx, func() {
+			for _, file := range files {
+				_ = file.SetReadDeadline(time.Now())
+			}
+		})
+		for i, file := range files {
+			var e error
+			s.pumps[i], e = readRelay(file)
+			if e != nil {
+				return e
+			}
+		}
+		return nil
+	})
+	return err
+}
+
+func (r *Relay) splitOutcome(s *splitRelay, i int, err error) {
+	if errors.Is(err, io.EOF) {
+		s.ended[i] = true
+		if s.ended[0] && s.ended[1] {
+			s.once.Do(func() { close(s.done) })
+		}
+	} else if err != nil && !r.closeInterrupted(err) {
+		r.err = errors.Join(r.err, err)
+		r.w.channels.fail(err)
+		r.cancel()
+	}
+}
+
+func (r *Relay) splitWriteOutcome(err error) bool {
+	if err == nil {
+		return true
+	}
+	if !r.closeInterrupted(err) && !errors.Is(err, errWorkInterrupted) {
+		r.err = errors.Join(r.err, err)
+		r.w.channels.fail(err)
+		r.cancel()
+	}
+	return false
+}
+
+// A split source may resume only after its complete admitted chunk reached the
+// terminal writer. An interrupted write can still have an empty suffix when
+// the boundary arrived after the final Write call; keep that source live in
+// the same way as an uninterrupted write. A nonempty suffix remains parked.
+func (r *Relay) resumeSplitIfReady(i int) {
+	s := r.split
+	if s == nil || s.pumps[i] == nil || len(r.w.pendingSplit[i]) != 0 || r.ctx.Err() != nil || r.w.channels.Err() != nil {
+		return
+	}
+	select {
+	case s.pumps[i].resume <- struct{}{}:
+	default:
+	}
+}
+
+func (r *Relay) consumeSplit(s *splitRelay, i int) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if s != r.split || r.ctx.Err() != nil || s.pumps[i] == nil {
+		return
+	}
+	p := s.pumps[i]
+	data, err := p.take()
+	writeErr := r.writeOutputStream(r.w.channels.ctx, data, splitStreams[i])
+	written := r.splitWriteOutcome(writeErr)
+	r.splitOutcome(s, i, err)
+	if err != nil {
+		<-p.done
+		s.pumps[i] = nil
+	} else if written || errors.Is(writeErr, errWorkInterrupted) {
+		r.resumeSplitIfReady(i)
+	}
+}
+
+// A source with an undelivered suffix stays stopped. The existing boundary's
+// wake resumes it only after that suffix reaches the sole terminal writer.
+func (r *Relay) flushSplitPending(ctx context.Context) error {
+	for i, stream := range splitStreams {
+		data := r.w.pendingSplit[i]
+		if len(data) == 0 {
+			continue
+		}
+		r.w.pendingSplit[i] = nil
+		err := r.writeOutputStream(ctx, data, stream)
+		written := r.splitWriteOutcome(err)
+		if written || errors.Is(err, errWorkInterrupted) {
+			r.resumeSplitIfReady(i)
+		}
+		if !written {
+			return err
+		}
+	}
+	return nil
+}
+
+// Called with mu held after relay cancellation or inside its phase boundary.
+// It preserves admitted bytes; observing two EOFs is a separate caller proof.
+func (r *Relay) joinSplit() (err error) {
+	s := r.split
+	if s == nil {
+		return nil
+	}
+	for i, p := range s.pumps {
+		if p != nil {
+			err = errors.Join(err, p.join(s.files[i]))
+			data, e := p.take()
+			r.w.pendingSplit[i] = append(r.w.pendingSplit[i], data...)
+			r.splitOutcome(s, i, e)
+			s.pumps[i] = nil
+		}
+	}
+	s.stopIO()
+	for i, file := range s.files {
+		if file != nil {
+			err = errors.Join(err, file.Close())
+			s.files[i] = nil
+		}
+	}
+	return err
+}
+
+// stopSplit joins only pump-owned I/O. The setup owner still owns its raw
+// readers, writer keepalives and paths. This is not a cleanup/EOF barrier.
+func (r *Relay) stopSplit(ctx context.Context) error {
+	_, err := r.boundary(ctx, false, func(int64) error {
+		if e := r.joinSplit(); e != nil {
+			return e
+		}
+		if e := r.flushSplitPending(ctx); e != nil {
+			return e
+		}
+		if r.err != nil {
+			return r.err
+		}
+		r.split = nil
+		return r.w.mark(ctx, true)
+	})
+	return err
+}
diff --git a/runtime/envoy/session/split_relay_linux_test.go b/runtime/envoy/session/split_relay_linux_test.go
new file mode 100644
index 00000000..564cad38
--- /dev/null
+++ b/runtime/envoy/session/split_relay_linux_test.go
@@ -0,0 +1,884 @@
+//go:build linux
+
+package session
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"runtime"
+	"sync"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/omry/omegaflow/runtime/envoy/protocol"
+)
+
+func splitRelayFor(t *testing.T) (relayFixture, *splitTestOwner, [2]*os.File) {
+	t.Helper()
+	f := relayFixtureFor(t)
+	s := splitOwnerFor(t)
+	ctx := splitContext(t)
+	if _, _, err := s.Prepare(ctx, splitRequest()); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		_ = f.r.Close(splitContext(t))
+		if err := s.Rollback(ctx, "operation-1"); err != nil {
+			t.Error(err)
+		}
+	})
+	if err := f.r.attachSplit(ctx, s.SplitResources, "operation-1"); err != nil {
+		t.Fatal(err)
+	}
+	var writers [2]*os.File
+	for i, stream := range splitStreams {
+		file, err := os.OpenFile(filepath.Join(s.root, "split", "operation-1."+stream), os.O_WRONLY, 0)
+		if err != nil {
+			t.Fatal(err)
+		}
+		writers[i] = file
+		t.Cleanup(func() { _ = file.Close() })
+	}
+	return f, s, writers
+}
+
+func splitRelayManualFor(t *testing.T) (relayFixture, *splitTestOwner, [2]*os.File) {
+	t.Helper()
+	f := relayFixtureFor(t)
+	s := splitOwnerFor(t)
+	ctx := splitContext(t)
+	if _, _, err := s.Prepare(ctx, splitRequest()); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		_ = f.r.Close(splitContext(t))
+		if err := s.Rollback(ctx, "operation-1"); err != nil {
+			t.Error(err)
+		}
+	})
+	files, err := s.relayReaders("operation-1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	attached := &splitRelay{id: "operation-1", files: files, done: make(chan struct{})}
+	attached.stopIO = joinedAfterFunc(f.r.ctx, func() {
+		for _, file := range files {
+			_ = file.SetReadDeadline(time.Now())
+		}
+	})
+	for i, file := range files {
+		attached.pumps[i], err = readRelay(file)
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+	f.r.mu.Lock()
+	f.r.split = attached
+	f.r.mu.Unlock()
+	select {
+	case f.r.wake <- struct{}{}:
+	default:
+	}
+	var writers [2]*os.File
+	for i, stream := range splitStreams {
+		file, err := os.OpenFile(filepath.Join(s.root, "split", "operation-1."+stream), os.O_WRONLY, 0)
+		if err != nil {
+			t.Fatal(err)
+		}
+		writers[i] = file
+		t.Cleanup(func() { _ = file.Close() })
+	}
+	return f, s, writers
+}
+
+func splitRead(t *testing.T, f relayFixture, n int) []byte {
+	t.Helper()
+	data := make([]byte, n)
+	_ = f.terminal.SetReadDeadline(time.Now().Add(3 * time.Second))
+	if _, err := io.ReadFull(f.terminal, data); err != nil {
+		t.Fatal("raw output", n, err)
+	}
+	return data
+}
+
+// The temporary focused overlay calls this seam immediately after a complete
+// terminal write and before writeOutputStream samples its context. The normal
+// source has no call site; the test skips there and the overlay run supplies
+// the deterministic scheduling boundary needed to prove the interruption.
+type splitWriteSeam struct {
+	stream      string
+	entered     chan struct{}
+	release     chan struct{}
+	enteredOnce sync.Once
+	releaseOnce sync.Once
+}
+
+var splitWriteSeamState struct {
+	sync.Mutex
+	current *splitWriteSeam
+}
+
+func armSplitWriteSeam(stream string) *splitWriteSeam {
+	s := &splitWriteSeam{stream: stream, entered: make(chan struct{}), release: make(chan struct{})}
+	splitWriteSeamState.Lock()
+	splitWriteSeamState.current = s
+	splitWriteSeamState.Unlock()
+	return s
+}
+
+func (s *splitWriteSeam) disarm() {
+	splitWriteSeamState.Lock()
+	if splitWriteSeamState.current == s {
+		splitWriteSeamState.current = nil
+	}
+	splitWriteSeamState.Unlock()
+	s.releaseOnce.Do(func() { close(s.release) })
+}
+
+func splitAfterCompletedStreamWrite(_ *Relay, stream string, ctx context.Context) {
+	splitWriteSeamState.Lock()
+	seam := splitWriteSeamState.current
+	if seam == nil || seam.stream != stream {
+		splitWriteSeamState.Unlock()
+		return
+	}
+	block := false
+	seam.enteredOnce.Do(func() {
+		close(seam.entered)
+		block = true
+	})
+	splitWriteSeamState.Unlock()
+	if !block {
+		return
+	}
+	<-seam.release
+	<-ctx.Done()
+}
+
+func waitSplitWriteSeam(t *testing.T, s *splitWriteSeam) bool {
+	t.Helper()
+	timer := time.NewTimer(500 * time.Millisecond)
+	defer timer.Stop()
+	select {
+	case <-s.entered:
+		return true
+	case <-timer.C:
+		return false
+	}
+}
+
+func closeSplitSources(s *splitTestOwner, writers [2]*os.File) {
+	for i, writer := range writers {
+		_ = writer.Close()
+		if s.operation.keepalives[i] >= 0 {
+			_ = syscall.Close(s.operation.keepalives[i])
+			s.operation.keepalives[i] = -1
+		}
+	}
+}
+
+// Interpret only the public mark stream over the raw log. There is no second
+// production log or per-stream byte counter to serve as the reconstruction.
+func splitReconstruct(t *testing.T, f relayFixture, raw []byte) map[string][]byte {
+	t.Helper()
+	result := map[string][]byte{}
+	stream, offset, elapsed := "pty", int64(0), int64(0)
+	for {
+		mark, ok := nextRelayFrame(t, f).(*protocol.OutputMark)
+		if !ok || mark.Offset < offset || mark.Offset > int64(len(raw)) || mark.ElapsedUS < elapsed {
+			t.Fatal("invalid stream frontier", mark, offset, len(raw))
+		}
+		result[stream] = append(result[stream], raw[offset:mark.Offset]...)
+		stream, offset, elapsed = mark.Stream, mark.Offset, mark.ElapsedUS
+		if offset == int64(len(raw)) {
+			return result
+		}
+	}
+}
+
+func TestSplitRelayExactLogicalBytesAndFirstSource(t *testing.T) {
+	for _, first := range []int{0, 1} {
+		t.Run(splitStreams[first], func(t *testing.T) {
+			f, _, writers := splitRelayFor(t)
+			var raw []byte
+			expected := map[string][]byte{}
+			for _, i := range []int{first, 1 - first, first, 2, 1 - first} {
+				data := []byte{'x', 0, 0xff, '\r', '\n', '\x1b', 3}
+				data = append(data, []byte("éλ終")...)
+				file, stream := f.slave, "pty"
+				if i < 2 {
+					file, stream = writers[i], splitStreams[i]
+				}
+				if _, err := file.Write(data); err != nil {
+					t.Fatal(err)
+				}
+				raw = append(raw, splitRead(t, f, len(data))...)
+				expected[stream] = append(expected[stream], data...)
+			}
+			if err := f.r.stopSplit(splitContext(t)); err != nil {
+				t.Fatal(err)
+			}
+			actual := splitReconstruct(t, f, raw)
+			for stream, want := range expected {
+				if !bytes.Equal(actual[stream], want) {
+					t.Fatalf("%s changed: %x != %x", stream, actual[stream], want)
+				}
+			}
+			// An empty boundary repeats the most recently written source.
+			if _, err := f.r.Drain(splitContext(t)); err != nil {
+				t.Fatal(err)
+			}
+			mark := nextRelayFrame(t, f).(*protocol.OutputMark)
+			if mark.Stream != splitStreams[1-first] || mark.Offset != int64(len(raw)) {
+				t.Fatal("empty boundary changed current source", mark)
+			}
+		})
+	}
+}
+
+func TestSplitRelayZeroBytesAndIndependentEOF(t *testing.T) {
+	f, s, writers := splitRelayFor(t)
+	f.r.mu.Lock()
+	attached := f.r.split
+	f.r.mu.Unlock()
+	for i := range writers {
+		_ = writers[i].Close()
+		if err := syscall.Close(s.operation.keepalives[i]); err != nil {
+			t.Fatal(err)
+		}
+		s.operation.keepalives[i] = -1
+		if i == 0 {
+			wait := splitContext(t)
+			for {
+				f.r.mu.Lock()
+				ended := attached.ended[0]
+				f.r.mu.Unlock()
+				if ended {
+					break
+				}
+				if wait.Err() != nil {
+					t.Fatal("first source EOF was not observed")
+				}
+				runtime.Gosched()
+			}
+			// The other keepalive must prevent a false two-EOF assertion.
+			select {
+			case <-attached.done:
+				t.Fatal("one closed source completed both EOF facts")
+			default:
+			}
+		}
+	}
+	select {
+	case <-attached.done:
+	case <-splitContext(t).Done():
+		t.Fatal("independent FIFO EOFs not observed")
+	}
+	if err := f.r.stopSplit(splitContext(t)); err != nil {
+		t.Fatal(err)
+	}
+	// Attachment and detachment both cover zero bytes, still attributed to PTY.
+	for range 3 {
+		m := nextRelayFrame(t, f).(*protocol.OutputMark)
+		if m.Offset != 0 || m.Stream != "pty" {
+			t.Fatal("zero-byte split invented a source", m)
+		}
+	}
+}
+
+func TestSplitRelayConcurrentLargeStreams(t *testing.T) {
+	f, _, writers := splitRelayFor(t)
+	want := [2][]byte{bytes.Repeat([]byte("stdout\x00\xff\n"), 30000), bytes.Repeat([]byte("stderr\r\nλ"), 30000)}
+	done := make(chan error, 2)
+	for i, file := range writers {
+		go func() { _, err := file.Write(want[i]); done <- err }()
+	}
+	raw := splitRead(t, f, len(want[0])+len(want[1]))
+	for range writers {
+		if err := <-done; err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := f.r.stopSplit(splitContext(t)); err != nil {
+		t.Fatal(err)
+	}
+	got := splitReconstruct(t, f, raw)
+	for i, stream := range splitStreams {
+		if !bytes.Equal(got[stream], want[i]) {
+			t.Fatal("concurrent logical stream changed", stream, len(got[stream]), len(want[i]))
+		}
+	}
+}
+
+func TestSplitRelayReaderIdentityAndOwnedDisposal(t *testing.T) {
+	f, s, _ := splitRelayFor(t)
+	f.r.mu.Lock()
+	var owned []int
+	var pumps []*relayReader
+	for i, file := range f.r.split.files {
+		raw, err := file.SyscallConn()
+		if err != nil {
+			f.r.mu.Unlock()
+			t.Fatal(err)
+		}
+		fd := -1
+		if err := raw.Control(func(value uintptr) { fd = int(value) }); err != nil {
+			f.r.mu.Unlock()
+			t.Fatal(err)
+		}
+		owned = append(owned, fd)
+		pumps = append(pumps, f.r.split.pumps[i])
+		flags, _, errno := syscall.Syscall(syscall.SYS_FCNTL, uintptr(fd), syscall.F_GETFD, 0)
+		if errno != 0 || flags&syscall.FD_CLOEXEC == 0 || fd == s.operation.readers[i] {
+			f.r.mu.Unlock()
+			t.Fatal("reader ownership/isolation", fd, flags, errno)
+		}
+	}
+	f.r.mu.Unlock()
+	if err := f.r.stopSplit(splitContext(t)); err != nil {
+		t.Fatal(err)
+	}
+	splitAssertClosed(t, owned)
+	for _, pump := range pumps {
+		select {
+		case <-pump.done:
+		default:
+			t.Fatal("retained pump")
+		}
+	}
+	if err := s.validate(); err != nil {
+		t.Fatal("detachment consumed setup-owned handles", err)
+	}
+	if err := f.r.attachSplit(splitContext(t), s.SplitResources, "other"); err == nil {
+		t.Fatal("wrong operation admitted")
+	}
+	ctx, cancel := context.WithCancel(splitContext(t))
+	cancel()
+	if err := f.r.attachSplit(ctx, s.SplitResources, "operation-1"); !errors.Is(err, context.Canceled) {
+		t.Fatal("canceled attachment admitted", err)
+	}
+	if err := os.Chmod(filepath.Join(s.root, "split", "operation-1.stdout"), 0644); err != nil {
+		t.Fatal(err)
+	}
+	if err := f.r.attachSplit(splitContext(t), s.SplitResources, "operation-1"); err == nil {
+		t.Fatal("changed reader identity admitted")
+	}
+	if err := os.Chmod(filepath.Join(s.root, "split", "operation-1.stdout"), 0600); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestSplitRelayAdmittedChunksRemainBoundedAndSurviveClose(t *testing.T) {
+	f, _, writers := splitRelayFor(t)
+	f.r.mu.Lock()
+	s := f.r.split
+	for i, file := range writers {
+		if _, err := file.Write(bytes.Repeat([]byte{byte('a' + i)}, 4096)); err != nil {
+			f.r.mu.Unlock()
+			t.Fatal(err)
+		}
+	}
+	ctx := splitContext(t)
+	for _, p := range s.pumps {
+		for {
+			p.mu.Lock()
+			n := len(p.data)
+			p.mu.Unlock()
+			if n > 0 {
+				if n > 32768 {
+					t.Fatal("unbounded reader", n)
+				}
+				break
+			}
+			if ctx.Err() != nil {
+				f.r.mu.Unlock()
+				t.Fatal("no admitted chunk")
+			}
+			runtime.Gosched()
+		}
+	}
+	closed := make(chan error, 1)
+	go func() { closed <- f.r.Close(ctx) }()
+	<-f.r.ctx.Done()
+	f.r.mu.Unlock()
+	if err := <-closed; err != nil {
+		t.Fatal(err)
+	}
+	for i, pending := range f.r.w.pendingSplit {
+		if !bytes.Equal(pending, bytes.Repeat([]byte{byte('a' + i)}, 4096)) {
+			t.Fatal("lost admitted stream", i, len(pending))
+		}
+	}
+	_ = f.slave.Close()
+	_ = f.r.w.PTY.SetReadDeadline(time.Now().Add(time.Second))
+	if err := f.r.w.drainTerminal(ctx); err != nil {
+		t.Fatal(err)
+	}
+	got := splitReconstruct(t, f, splitRead(t, f, 8192))
+	for i, stream := range splitStreams {
+		if !bytes.Equal(got[stream], bytes.Repeat([]byte{byte('a' + i)}, 4096)) {
+			t.Fatal("final drain misattributed retained output", stream)
+		}
+	}
+}
+
+func TestSplitRelayReadFailureFailsSession(t *testing.T) {
+	f, _, _ := splitRelayFor(t)
+	f.r.mu.Lock()
+	file := f.r.split.files[0]
+	f.r.mu.Unlock()
+	if err := file.Close(); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-f.r.Done():
+	case <-splitContext(t).Done():
+		t.Fatal("read failure did not join the relay")
+	}
+	if f.r.Err() == nil || f.r.w.channels.Err() == nil {
+		t.Fatal("read failure produced success")
+	}
+}
+
+func TestSplitRelayCadenceAndExistingSessionBudgets(t *testing.T) {
+	t.Run("cadence", func(t *testing.T) {
+		f, _, writers := splitRelayFor(t)
+		if _, err := writers[0].Write([]byte("first")); err != nil {
+			t.Fatal(err)
+		}
+		_ = splitRead(t, f, 5)
+		<-time.NewTimer(markCadence).C
+		if _, err := writers[0].Write([]byte("second")); err != nil {
+			t.Fatal(err)
+		}
+		_ = splitRead(t, f, 6)
+		if err := f.r.stopSplit(splitContext(t)); err != nil {
+			t.Fatal(err)
+		}
+		seen := false
+		for {
+			mark := nextRelayFrame(t, f).(*protocol.OutputMark)
+			if mark.Offset == 5 && mark.Stream == "stdout" {
+				seen = true
+			}
+			if mark.Offset == 11 {
+				break
+			}
+		}
+		if !seen {
+			t.Fatal("same-stream sender cadence mark omitted")
+		}
+	})
+	for _, kind := range []string{"marks", "bytes"} {
+		t.Run(kind, func(t *testing.T) {
+			f, _, writers := splitRelayFor(t)
+			f.r.mu.Lock()
+			if kind == "marks" {
+				f.r.w.marks = markLimit
+			} else {
+				f.r.w.OutputThrough = outputLimit
+			}
+			before := f.r.w.OutputThrough
+			f.r.mu.Unlock()
+			if _, err := writers[1].Write([]byte("over budget")); err != nil {
+				t.Fatal(err)
+			}
+			select {
+			case <-f.r.Done():
+			case <-splitContext(t).Done():
+				t.Fatal("budget crossing did not fail relay")
+			}
+			if f.r.Err() == nil || f.r.w.OutputThrough != before {
+				t.Fatal("split source reset or crossed session budget", f.r.Err(), f.r.w.OutputThrough)
+			}
+		})
+	}
+}
+
+func TestSplitRelayRejectsDuplicateAttachment(t *testing.T) {
+	f, s, _ := splitRelayFor(t)
+	f.r.mu.Lock()
+	owned := f.r.split
+	f.r.mu.Unlock()
+	if err := f.r.attachSplit(splitContext(t), s.SplitResources, "operation-1"); err == nil {
+		t.Fatal("two split read owners admitted")
+	}
+	select {
+	case <-f.r.Done():
+	case <-splitContext(t).Done():
+		t.Fatal("failed attachment did not join partial ownership")
+	}
+	for _, file := range owned.files {
+		if file != nil {
+			t.Fatal("duplicate failure retained original pump file")
+		}
+	}
+	if err := s.validate(); err != nil {
+		t.Fatal("failed attachment consumed setup owner", err)
+	}
+}
+
+func TestSplitRelayInterruptedWriterRetainsExactSourceSuffix(t *testing.T) {
+	for _, stream := range splitStreams {
+		t.Run(stream, func(t *testing.T) {
+			f := relayFixtureFor(t)
+			_ = f.r.w.channels.terminal.SetWriteBuffer(1024)
+			payload := bytes.Repeat([]byte("exact\x00\xff\r\n"), 300000)
+			ctx, cancel := context.WithCancelCause(splitContext(t))
+			defer cancel(nil)
+			done := make(chan error, 1)
+			go func() {
+				f.r.mu.Lock()
+				err := f.r.writeOutputStream(ctx, payload, stream)
+				f.r.mu.Unlock()
+				done <- err
+			}()
+			// The source mark proves the writer entered its actual write phase.
+			m := nextRelayFrame(t, f).(*protocol.OutputMark)
+			if m.Stream != stream || m.Offset != 0 {
+				t.Fatal(m)
+			}
+			cancel(errWorkInterrupted)
+			if err := <-done; !errors.Is(err, errWorkInterrupted) {
+				t.Fatal("missing interrupted writer result", err)
+			}
+			i := 0
+			if stream == "stderr" {
+				i = 1
+			}
+			f.r.mu.Lock()
+			through := f.r.w.OutputThrough
+			pending := append([]byte(nil), f.r.w.pendingSplit[i]...)
+			f.r.mu.Unlock()
+			if len(pending) == 0 || !bytes.Equal(pending, payload[through:]) {
+				t.Fatal("interrupted suffix lost source/bytes", through, len(pending))
+			}
+			read := make(chan error, 1)
+			go func() {
+				got := make([]byte, len(payload))
+				_ = f.terminal.SetReadDeadline(time.Now().Add(3 * time.Second))
+				_, err := io.ReadFull(f.terminal, got)
+				if err == nil && !bytes.Equal(got, payload) {
+					err = fmt.Errorf("interrupted bytes changed")
+				}
+				read <- err
+			}()
+			f.r.mu.Lock()
+			err := f.r.flushSplitPending(splitContext(t))
+			f.r.mu.Unlock()
+			if e := <-read; err != nil || e != nil {
+				t.Fatal("retry failed", err, e)
+			}
+		})
+	}
+}
+
+func TestSplitRelayPhaseRetryResumesLiveSource(t *testing.T) {
+	f, _, writers := splitRelayFor(t)
+	if err := f.terminal.SetReadBuffer(4096); err != nil {
+		t.Fatal(err)
+	}
+	if err := f.r.w.channels.terminal.SetWriteBuffer(1024); err != nil {
+		t.Fatal(err)
+	}
+	f.r.mu.Lock()
+	_ = f.r.w.channels.terminal.SetWriteDeadline(time.Now().Add(100 * time.Millisecond))
+	n, err := f.r.w.channels.terminal.Write(bytes.Repeat([]byte("p"), 1<<20))
+	if err == nil || n == 1<<20 {
+		f.r.mu.Unlock()
+		t.Fatal("backpressure not established", n, err)
+	}
+	f.r.w.OutputThrough += int64(n)
+	f.r.mu.Unlock()
+	first, second := bytes.Repeat([]byte("a\x00\xff\n"), 1024), []byte("following FIFO chunk\r\n")
+	if _, err := writers[0].Write(first); err != nil {
+		t.Fatal(err)
+	}
+	for {
+		m := nextRelayFrame(t, f).(*protocol.OutputMark)
+		if m.Stream == "stdout" {
+			break
+		}
+	}
+	job, interrupt := context.WithCancelCause(splitContext(t))
+	defer interrupt(nil)
+	boundary := make(chan error, 1)
+	go func() { _, err := f.r.boundary(job, false, nil); boundary <- err }()
+	wait := splitContext(t)
+	for {
+		f.r.phaseMu.Lock()
+		entered := f.r.phase == job
+		f.r.phaseMu.Unlock()
+		if entered {
+			break
+		}
+		if wait.Err() != nil {
+			t.Fatal("boundary did not acquire the blocked writer's epoch")
+		}
+		runtime.Gosched()
+	}
+	interrupt(errWorkInterrupted)
+	select {
+	case err := <-boundary:
+		if !errors.Is(err, context.Canceled) {
+			t.Fatal("wrong interruption", err)
+		}
+	case <-wait.Done():
+		t.Fatal("boundary did not release")
+	}
+	if f.r.w.channels.Err() != nil {
+		t.Fatal("phase interruption failed session", f.r.w.channels.Err())
+	}
+	want := append(bytes.Repeat([]byte("p"), n), first...)
+	want = append(want, second...)
+	read := make(chan error, 1)
+	go func() {
+		_ = f.terminal.SetReadDeadline(time.Now().Add(3 * time.Second))
+		got := make([]byte, len(want))
+		_, err := io.ReadFull(f.terminal, got)
+		if err == nil && !bytes.Equal(got, want) {
+			err = fmt.Errorf("live source changed after retry")
+		}
+		read <- err
+	}()
+	if _, err := writers[0].Write(second); err != nil {
+		t.Fatal(err)
+	}
+	if err := <-read; err != nil {
+		t.Fatal("live source did not resume after its retained suffix", err)
+	}
+	if err := f.r.stopSplit(splitContext(t)); err != nil {
+		t.Fatal(err)
+	}
+	if f.r.w.OutputThrough != int64(len(want)) || len(f.r.w.pendingSplit[0]) != 0 {
+		t.Fatal("retry lost or duplicated bytes")
+	}
+}
+
+func TestSplitRelayCompletedInitialChunkInterruptionResumesLiveSource(t *testing.T) {
+	for source, stream := range splitStreams {
+		t.Run(stream, func(t *testing.T) {
+			seam := armSplitWriteSeam(stream)
+			t.Cleanup(seam.disarm)
+			f, s, writers := splitRelayFor(t)
+			first, second := []byte("first-complete-chunk"), []byte("next-fifo-chunk")
+			if _, err := writers[source].Write(first); err != nil {
+				t.Fatal(err)
+			}
+			if !waitSplitWriteSeam(t, seam) {
+				t.Skip("requires the temporary completed-write scheduling overlay")
+			}
+			if got := splitRead(t, f, len(first)); !bytes.Equal(got, first) {
+				t.Fatal("initial bytes changed", got)
+			}
+
+			job, interrupt := context.WithCancelCause(splitContext(t))
+			defer interrupt(nil)
+			boundary := make(chan error, 1)
+			go func() { _, err := f.r.boundary(job, false, nil); boundary <- err }()
+			wait := splitContext(t)
+			for {
+				f.r.phaseMu.Lock()
+				entered := f.r.phase == job
+				f.r.phaseMu.Unlock()
+				if entered {
+					break
+				}
+				if wait.Err() != nil {
+					t.Fatal("boundary did not acquire the completed writer epoch")
+				}
+				runtime.Gosched()
+			}
+			interrupt(errWorkInterrupted)
+			seam.releaseOnce.Do(func() { close(seam.release) })
+			select {
+			case err := <-boundary:
+				if !errors.Is(err, context.Canceled) {
+					t.Fatal("wrong interruption", err)
+				}
+			case <-wait.Done():
+				t.Fatal("boundary did not release")
+			}
+
+			f.r.mu.Lock()
+			attached := f.r.split
+			pending, through := len(f.r.w.pendingSplit[source]), f.r.w.OutputThrough
+			f.r.mu.Unlock()
+			if pending != 0 || through != int64(len(first)) {
+				t.Fatalf("complete interrupted chunk was not acknowledged: pending=%d through=%d", pending, through)
+			}
+			if f.r.w.channels.Err() != nil {
+				t.Fatal("phase interruption failed session", f.r.w.channels.Err())
+			}
+			if _, err := writers[source].Write(second); err != nil {
+				t.Fatal(err)
+			}
+			if got := splitRead(t, f, len(second)); !bytes.Equal(got, second) {
+				t.Fatal("resumed source bytes changed", got)
+			}
+			closeSplitSources(s, writers)
+			select {
+			case <-attached.done:
+			case <-splitContext(t).Done():
+				f.r.mu.Lock()
+				ended := attached.ended
+				pumps := [2]bool{attached.pumps[0] != nil, attached.pumps[1] != nil}
+				f.r.mu.Unlock()
+				t.Fatalf("independent FIFO EOFs did not arrive: ended=%v pumps=%v channels=%v relay=%v", ended, pumps, f.r.w.channels.Err(), f.r.Err())
+			}
+			if err := f.r.stopSplit(splitContext(t)); err != nil {
+				t.Fatal(err)
+			}
+			if f.r.w.OutputThrough != int64(len(first)+len(second)) || len(f.r.w.pendingSplit[source]) != 0 {
+				t.Fatal("resumed source duplicated or retained bytes")
+			}
+		})
+	}
+}
+
+func TestSplitRelayCompletedSuffixInterruptionResumesLiveSource(t *testing.T) {
+	for source, stream := range splitStreams {
+		t.Run(stream, func(t *testing.T) {
+			f, s, writers := splitRelayManualFor(t)
+			if err := f.r.w.channels.terminal.SetWriteBuffer(1024); err != nil {
+				t.Fatal(err)
+			}
+			prefix := bytes.Repeat([]byte("p"), 1<<20)
+			f.r.mu.Lock()
+			_ = f.r.w.channels.terminal.SetWriteDeadline(time.Now().Add(100 * time.Millisecond))
+			n, err := f.r.w.channels.terminal.Write(prefix)
+			if err == nil || n == len(prefix) {
+				f.r.mu.Unlock()
+				t.Fatal("backpressure not established", n, err)
+			}
+			f.r.w.OutputThrough += int64(n)
+			f.r.mu.Unlock()
+			first := bytes.Repeat([]byte("source\x00\xff\r\n"), 3000)
+			job, interrupt := context.WithCancelCause(splitContext(t))
+			defer interrupt(nil)
+			f.r.phaseMu.Lock()
+			f.r.phase = job
+			f.r.phaseMu.Unlock()
+			t.Cleanup(func() {
+				f.r.phaseMu.Lock()
+				f.r.phase = nil
+				f.r.phaseMu.Unlock()
+			})
+			if _, err := writers[source].Write(first); err != nil {
+				t.Fatal(err)
+			}
+			for {
+				mark := nextRelayFrame(t, f).(*protocol.OutputMark)
+				if mark.Stream == stream {
+					break
+				}
+			}
+			interrupt(errWorkInterrupted)
+			wait := splitContext(t)
+			var through int64
+			var pending []byte
+			for {
+				f.r.mu.Lock()
+				through = f.r.w.OutputThrough
+				pending = append([]byte(nil), f.r.w.pendingSplit[source]...)
+				f.r.mu.Unlock()
+				f.r.phaseMu.Lock()
+				writerDone := f.r.writerCancel == nil
+				f.r.phaseMu.Unlock()
+				if writerDone && len(pending) != 0 {
+					break
+				}
+				if wait.Err() != nil {
+					t.Fatal("interrupted source did not retain a suffix")
+				}
+				runtime.Gosched()
+			}
+			f.r.phaseMu.Lock()
+			f.r.phase = nil
+			f.r.phaseMu.Unlock()
+			before := int64(n)
+			if through < before || through-before > int64(len(first)) || !bytes.Equal(pending, first[through-before:]) {
+				t.Fatal("nonempty retained suffix was not preserved", through, len(pending))
+			}
+
+			seam := armSplitWriteSeam(stream)
+			t.Cleanup(seam.disarm)
+			retry, retryCancel := context.WithCancelCause(splitContext(t))
+			defer retryCancel(nil)
+			flushed := make(chan error, 1)
+			go func() {
+				f.r.mu.Lock()
+				err := f.r.flushSplitPending(retry)
+				f.r.mu.Unlock()
+				flushed <- err
+			}()
+			read := make(chan struct {
+				data []byte
+				err  error
+			}, 1)
+			go func() {
+				data := make([]byte, n+len(first))
+				_ = f.terminal.SetReadDeadline(time.Now().Add(3 * time.Second))
+				_, err := io.ReadFull(f.terminal, data)
+				read <- struct {
+					data []byte
+					err  error
+				}{data, err}
+			}()
+			if !waitSplitWriteSeam(t, seam) {
+				t.Skip("requires the temporary completed-write scheduling overlay")
+			}
+			result := <-read
+			want := append(prefix[:n], first...)
+			if result.err != nil || !bytes.Equal(result.data, want) {
+				t.Fatal("retained suffix bytes changed", result.err)
+			}
+			retryCancel(errWorkInterrupted)
+			seam.releaseOnce.Do(func() { close(seam.release) })
+			if err := <-flushed; !errors.Is(err, errWorkInterrupted) {
+				t.Fatal("missing interrupted flush result", err)
+			}
+
+			f.r.mu.Lock()
+			attached := f.r.split
+			through = f.r.w.OutputThrough
+			pendingLen := len(f.r.w.pendingSplit[source])
+			f.r.mu.Unlock()
+			if pendingLen != 0 || through != int64(n+len(first)) {
+				t.Fatalf("complete interrupted suffix was not acknowledged: pending=%d through=%d", pendingLen, through)
+			}
+			if f.r.w.channels.Err() != nil {
+				t.Fatal("suffix interruption failed session", f.r.w.channels.Err())
+			}
+			following := []byte("following-fifo-chunk")
+			if _, err := writers[source].Write(following); err != nil {
+				t.Fatal(err)
+			}
+			if got := splitRead(t, f, len(following)); !bytes.Equal(got, following) {
+				t.Fatal("resumed suffix source bytes changed", got)
+			}
+			closeSplitSources(s, writers)
+			select {
+			case <-attached.done:
+			case <-splitContext(t).Done():
+				f.r.mu.Lock()
+				ended := attached.ended
+				pumps := [2]bool{attached.pumps[0] != nil, attached.pumps[1] != nil}
+				f.r.mu.Unlock()
+				t.Fatalf("independent FIFO EOFs did not arrive after suffix retry: ended=%v pumps=%v channels=%v relay=%v", ended, pumps, f.r.w.channels.Err(), f.r.Err())
+			}
+			if err := f.r.stopSplit(splitContext(t)); err != nil {
+				t.Fatal(err)
+			}
+			if f.r.w.OutputThrough != int64(n+len(first)+len(following)) || len(f.r.w.pendingSplit[source]) != 0 {
+				t.Fatal("suffix retry duplicated or retained bytes")
+			}
+		})
+	}
+}
diff --git a/runtime/envoy/session/split_relay_native_linux_test.go b/runtime/envoy/session/split_relay_native_linux_test.go
new file mode 100644
index 00000000..5a79fc92
--- /dev/null
+++ b/runtime/envoy/session/split_relay_native_linux_test.go
@@ -0,0 +1,219 @@
+//go:build linux
+
+package session
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"os"
+	"sync"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/omry/omegaflow/runtime/envoy/protocol"
+)
+
+// Attach the real B5.2 transport to the real B5.1 resources at the test-only
+// assembly boundary. Production start/completion integration belongs to B5.3.
+type nativeSplitTransport struct {
+	*SplitResources
+	r        *Relay
+	deadline time.Time
+}
+
+func (s *nativeSplitTransport) Prepare(ctx context.Context, req protocol.Execute) (string, string, error) {
+	s.deadline, _ = ctx.Deadline()
+	out, errout, err := s.SplitResources.Prepare(ctx, req)
+	if err == nil {
+		err = s.r.attachSplit(ctx, s.SplitResources, req.OperationID)
+	}
+	return out, errout, err
+}
+
+func TestNativeSplitStreamsExactBytes(t *testing.T) {
+	if os.Getenv("OMEGAFLOW_NATIVE_SPLIT_STREAMS") != "1" {
+		t.Skip("genuinely qualified isolated target required")
+	}
+	for _, first := range []string{"stdout", "stderr", "empty"} {
+		t.Run(first, func(t *testing.T) {
+			n := nativeOrdinaryFor(t)
+			owner := &nativeSplitTransport{SplitResources: NewSplitSetup(), r: n.c.r}
+			n.c.setup = owner
+			t.Cleanup(func() {
+				ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+				defer cancel()
+				_ = n.c.r.Close(ctx)
+				if owner.operation != nil {
+					original, end := context.WithDeadline(context.Background(), owner.deadline)
+					defer end()
+					if err := owner.Rollback(original, "operation-1"); err != nil {
+						t.Error("test-owned transport disposal", err)
+					}
+				}
+			})
+			req := splitRequest()
+			req.Header = protocol.Header{Schema: protocol.Schema, Type: "execute", Seq: n.seq}
+			req.Source = ":"
+			want := map[string][]byte{}
+			if first != "empty" {
+				redirect := ""
+				if first == "stderr" {
+					redirect = " >&2"
+				}
+				req.Source = "builtin printf 'first\\000\\377\\r\\n'" + redirect + "; for i in {1..2048}; do builtin printf 'out-é\\000\\377\\r\\n'; builtin printf 'err-λ\\000\\376\\n' >&2; done"
+				want["stdout"] = bytes.Repeat([]byte("out-é\x00\xff\r\n"), 2048)
+				want["stderr"] = bytes.Repeat([]byte("err-λ\x00\xfe\n"), 2048)
+				want[first] = append([]byte("first\x00\xff\r\n"), want[first]...)
+			}
+			n.request(t, &req)
+			e := receiveInput(t, n.c.w.channels)
+			actual, ok := e.Message.(*protocol.Execute)
+			if !ok {
+				t.Fatal(e)
+			}
+			// Drain the actual terminal concurrently, so a blocked test reader
+			// cannot substitute a small-output assumption for transport evidence.
+			type rawResult struct {
+				data []byte
+				err  error
+			}
+			rawDone := make(chan rawResult, 1)
+			sink := &nativeSplitSink{wake: make(chan struct{}, 1)}
+			_ = n.terminal.SetReadDeadline(time.Now().Add(5 * time.Second))
+			go func() {
+				_, err := io.Copy(sink, n.terminal)
+				rawDone <- rawResult{sink.bytes(), err}
+			}()
+			rawJoined := false
+			t.Cleanup(func() {
+				if !rawJoined {
+					_ = n.terminal.SetReadDeadline(time.Now())
+					select {
+					case <-rawDone:
+					case <-time.After(time.Second):
+						t.Error("test terminal collector did not join")
+					}
+				}
+			})
+			out, err := n.c.Start(n.c.w.channels.ctx, *actual)
+			if err != nil || !out.Started {
+				t.Fatal("qualified split source did not start", out, err)
+			}
+			// The helper's proposal proves source return only. The test does not
+			// release input_closed or claim an operation result/descendant proof.
+			select {
+			case result := <-n.c.private.events:
+				proposal, ok := result.message.(*protocol.PrivateInputClose)
+				if result.err != nil || !ok || proposal.OperationID != req.OperationID {
+					t.Fatal("actual source return", result)
+				}
+			case result := <-n.c.private.fatal:
+				t.Fatal(result)
+			case <-time.After(5 * time.Second):
+				t.Fatal("actual source did not return")
+			}
+			n.c.r.mu.Lock()
+			attached := n.c.r.split
+			n.c.r.mu.Unlock()
+			// Only this source and the setup keepalives can write these FIFOs.
+			// B5.3 must supply its real cleanup ordering before doing this.
+			for i, fd := range owner.operation.keepalives {
+				if err := syscall.Close(fd); err != nil {
+					t.Fatal(err)
+				}
+				owner.operation.keepalives[i] = -1
+			}
+			select {
+			case <-attached.done:
+			case <-time.After(3 * time.Second):
+				t.Fatal("actual FIFO streams did not end")
+			}
+			ctx := splitContext(t)
+			if err := n.c.r.stopSplit(ctx); err != nil {
+				t.Fatal(err)
+			}
+			through, err := n.c.r.Drain(ctx)
+			if err != nil {
+				t.Fatal(err)
+			}
+			for int64(len(sink.bytes())) < through-n.through {
+				select {
+				case <-sink.wake:
+				case <-ctx.Done():
+					t.Fatal("terminal did not receive covering range")
+				}
+			}
+			_ = n.terminal.SetReadDeadline(time.Now())
+			raw := <-rawDone
+			rawJoined = true
+			if !os.IsTimeout(raw.err) || int64(len(raw.data)) != through-n.through {
+				t.Fatal("raw native range", len(raw.data), through, n.through, raw.err)
+			}
+			got := map[string][]byte{}
+			stream, offset := "pty", n.through
+			started := false
+			for {
+				m := nativeFrame(t, n.reader)
+				encoded, err := protocol.EncodeEnvoy(m)
+				var h protocol.Header
+				if err != nil || json.Unmarshal(encoded, &h) != nil || h.Seq != n.lastSeq+1 {
+					t.Fatal("public sequence", m, err)
+				}
+				n.lastSeq = h.Seq
+				if mark, ok := m.(*protocol.OutputMark); ok {
+					if mark.Offset < offset || mark.Offset > through || mark.ElapsedUS < n.markElapsed {
+						t.Fatal("native mark frontier", mark)
+					}
+					got[stream] = append(got[stream], raw.data[offset-n.through:mark.Offset-n.through]...)
+					stream, offset, n.markElapsed = mark.Stream, mark.Offset, mark.ElapsedUS
+					if started && offset == through {
+						break
+					}
+				} else if event, ok := m.(*protocol.OperationStarted); !ok || event.OperationID != req.OperationID || event.OutputStart != out.OutputStart {
+					t.Fatal("unexpected native event", m)
+				} else {
+					started = true
+				}
+			}
+			for _, name := range splitStreams {
+				if !bytes.Equal(got[name], want[name]) {
+					t.Fatal("native logical bytes changed", name, len(got[name]), len(want[name]))
+				}
+			}
+			t.Logf("raw PTY controls: %q; exact stdout=%d stderr=%d", got["pty"], len(got["stdout"]), len(got["stderr"]))
+			// Readline's submit line ending is PTY traffic, outside both exact
+			// logical byte comparisons above. Keep its raw bytes in the evidence.
+			if visible := ordinaryVisible(string(got["pty"])); visible != "" && visible != "\n" {
+				t.Fatal("split source leaked into PTY", fmt.Sprintf("%q", visible))
+			}
+		})
+	}
+}
+
+// A test-only terminal collector reports actual read progress, so stopping it
+// cannot discard bytes that the writer sent but the socket has not delivered.
+type nativeSplitSink struct {
+	mu     sync.Mutex
+	buffer bytes.Buffer
+	wake   chan struct{}
+}
+
+func (s *nativeSplitSink) Write(p []byte) (int, error) {
+	s.mu.Lock()
+	n, err := s.buffer.Write(p)
+	s.mu.Unlock()
+	select {
+	case s.wake <- struct{}{}:
+	default:
+	}
+	return n, err
+}
+func (s *nativeSplitSink) bytes() []byte {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return append([]byte(nil), s.buffer.Bytes()...)
+}
diff --git a/runtime/envoy/session/startup_linux.go b/runtime/envoy/session/startup_linux.go
index 478ea322..9a871076 100644
--- a/runtime/envoy/session/startup_linux.go
+++ b/runtime/envoy/session/startup_linux.go
@@ -49,7 +49,9 @@ type Workload struct {
 	closeErr             error
 	relayOwned           atomic.Bool
 	privateOwned         atomic.Bool
-	pendingPTY           []byte // at most one relay chunk, transferred after join
+	pendingPTY           []byte    // at most one relay chunk, transferred after join
+	pendingSplit         [2][]byte // one admitted suffix per split source
+	outputStream         string
 	marks                int
 	lastMark             time.Duration
 }
@@ -486,23 +488,39 @@ func (w *Workload) verifyIdentity() error {
 }
 
 func (w *Workload) mark(ctx context.Context, force bool) error {
+	stream := w.outputStream
+	if stream == "" {
+		stream = "pty"
+	}
+	return w.markStream(ctx, stream, force)
+}
+
+func (w *Workload) markStream(ctx context.Context, stream string, force bool) error {
+	current := w.outputStream
+	if current == "" {
+		current = "pty"
+	}
+	if stream != "pty" && stream != "stdout" && stream != "stderr" {
+		return fmt.Errorf("invalid output stream")
+	}
 	elapsed := time.Since(w.ReadyAt)
 	if elapsed < w.lastMark {
 		return fmt.Errorf("output mark time regressed")
 	}
-	if !force && elapsed-w.lastMark < markCadence {
+	if !force && stream == current && elapsed-w.lastMark < markCadence {
 		return nil
 	}
 	if w.marks >= markLimit {
 		return fmt.Errorf("output mark budget exhausted")
 	}
 	if err := w.channels.sendEvent(ctx, "output_mark", func(h protocol.Header) protocol.Message {
-		return protocol.OutputMark{Header: h, Offset: w.OutputThrough, Stream: "pty", ElapsedUS: elapsed.Microseconds()}
+		return protocol.OutputMark{Header: h, Offset: w.OutputThrough, Stream: stream, ElapsedUS: elapsed.Microseconds()}
 	}); err != nil {
 		return err
 	}
 	w.marks++
 	w.lastMark = elapsed
+	w.outputStream = stream
 	return nil
 }
 
@@ -614,6 +632,24 @@ func (w *Workload) shutdownIdle(phase context.Context, reason string, next func(
 
 // Called only after the selected shell and direct Awsh child have been reaped.
 func (w *Workload) drainTerminal(ctx context.Context) error {
+	for i, stream := range splitStreams {
+		if len(w.pendingSplit[i]) > 0 {
+			if err := w.markStream(ctx, stream, false); err != nil {
+				return err
+			}
+			before := w.OutputThrough
+			err := w.writeTerminal(ctx, w.pendingSplit[i], true)
+			w.pendingSplit[i] = w.pendingSplit[i][w.OutputThrough-before:]
+			if err != nil {
+				return err
+			}
+		}
+	}
+	if len(w.pendingPTY) > 0 {
+		if err := w.markStream(ctx, "pty", false); err != nil {
+			return err
+		}
+	}
 	before := w.OutputThrough
 	err := w.writeTerminal(ctx, w.pendingPTY, true)
 	w.pendingPTY = w.pendingPTY[w.OutputThrough-before:]
@@ -624,6 +660,9 @@ func (w *Workload) drainTerminal(ctx context.Context) error {
 	for {
 		n, e := w.PTY.Read(buffer[:])
 		if n > 0 {
+			if err := w.markStream(ctx, "pty", false); err != nil {
+				return err
+			}
 			if err := w.writeTerminal(ctx, buffer[:n], true); err != nil {
 				return err
 			}
diff --git a/tests/envoy-split-streams/README.md b/tests/envoy-split-streams/README.md
new file mode 100644
index 00000000..1e8d9a56
--- /dev/null
+++ b/tests/envoy-split-streams/README.md
@@ -0,0 +1,41 @@
+# Split stream transport proof (B5.2)
+
+Run the portable isolated transport tests:
+
+```sh
+cd runtime/envoy
+go test -buildvcs=false -race ./session -run '^TestSplitRelay' -count=1
+```
+
+Run actual Bash stream proofs from the repository root:
+
+```sh
+python3 tests/envoy-split-streams/run.py --qualification QUALIFICATION --output NEW_DIRECTORY
+```
+
+The qualifier must contain genuine complete receipts for the three pinned
+Linux amd64 targets. The runner checks exact Awsh and startup asset hashes,
+stages the same session test binary in each immutable target, and retains
+image identities, binary hashes, commands and native logs. Rebuildable binaries
+are removed on success and failure. ARM64 compilation is not qualification.
+
+Each stream uses its own real FIFO and one bounded reader. Both feed the
+existing sole terminal writer. Public output marks over the single global raw
+offset reconstruct exact logical stdout and stderr. Tests compare binary,
+invalid UTF-8, NUL, CR/LF and multibyte UTF-8 without normalization. Only the
+known Readline controls in the separate PTY stream receive a diagnostic check;
+the raw PTY bytes remain in the native log.
+
+Tests cover stdout-first and stderr-first output, concurrent large streams,
+PTY interleaving, empty boundaries, independent EOF facts, reader identity and
+close-on-exec isolation, joined disposal, source failure, bounded admission,
+interrupted writes and exact attributed suffix recovery. An EOF fact alone does
+not constitute an operation result or a cleanup receipt.
+
+The native test uses an explicit test-only attachment of the real setup and
+relay primitives. It observes actual source return, then closes the test-owned
+writer keepalives and joins the transport. It does not send input_closed or
+claim completed split operation semantics. B5.3 owns production start/resize/
+completion integration and the cleanup-before-keepalive, dual-EOF/removal/result
+barriers. B6 owns real descendant cleanup. The case map preserves these
+cumulative owners; no approved corpus file is rewritten here.
diff --git a/tests/envoy-split-streams/case-map.json b/tests/envoy-split-streams/case-map.json
new file mode 100644
index 00000000..76cbaeeb
--- /dev/null
+++ b/tests/envoy-split-streams/case-map.json
@@ -0,0 +1,60 @@
+{
+  "slice": "B5.2",
+  "cases": [
+    {
+      "inventory_id": "B1-C005-pty-start",
+      "test": "TestSplitRelayExactLogicalBytesAndFirstSource",
+      "proof": "Real FIFO/PTY source switches at one raw offset; empty boundaries repeat the current source",
+      "cumulative_owner": "B5.3",
+      "cumulative_closeout": false
+    },
+    {
+      "inventory_id": "B1-C005-pty-repeat-current-stream",
+      "test": "TestSplitRelayExactLogicalBytesAndFirstSource",
+      "proof": "Real FIFO/PTY source switches at one raw offset; empty boundaries repeat the current source",
+      "cumulative_owner": "B5.3",
+      "cumulative_closeout": false
+    },
+    {
+      "inventory_id": "B1-C005-stdout-start",
+      "test": "TestSplitRelayExactLogicalBytesAndFirstSource",
+      "proof": "Real FIFO/PTY source switches at one raw offset; empty boundaries repeat the current source",
+      "cumulative_owner": "B5.3",
+      "cumulative_closeout": false
+    },
+    {
+      "inventory_id": "B1-C005-stdout-repeat-current-stream",
+      "test": "TestSplitRelayExactLogicalBytesAndFirstSource",
+      "proof": "Real FIFO/PTY source switches at one raw offset; empty boundaries repeat the current source",
+      "cumulative_owner": "B5.3",
+      "cumulative_closeout": false
+    },
+    {
+      "inventory_id": "B1-C005-stderr-start",
+      "test": "TestSplitRelayExactLogicalBytesAndFirstSource",
+      "proof": "Real FIFO/PTY source switches at one raw offset; empty boundaries repeat the current source",
+      "cumulative_owner": "B5.3",
+      "cumulative_closeout": false
+    },
+    {
+      "inventory_id": "B1-C005-stderr-repeat-current-stream",
+      "test": "TestSplitRelayExactLogicalBytesAndFirstSource",
+      "proof": "Real FIFO/PTY source switches at one raw offset; empty boundaries repeat the current source",
+      "cumulative_owner": "B5.3",
+      "cumulative_closeout": false
+    },
+    {
+      "inventory_id": "B1-C034-interleaved-logical-bytes",
+      "test": "TestSplitRelayConcurrentLargeStreams",
+      "proof": "Exact binary and UTF-8 stdout/stderr reconstructed solely from public marks and one raw terminal log",
+      "cumulative_owner": "B6.3",
+      "cumulative_closeout": false
+    },
+    {
+      "inventory_id": "B1-PACKAGE-B5.2",
+      "test": "TestNativeSplitStreamsExactBytes",
+      "proof": "Actual qualified Bash writes exact independent FIFO streams; stdout-first, stderr-first and zero-byte cases",
+      "qualification": "genuine pinned Linux amd64 targets only"
+    }
+  ]
+}
diff --git a/tests/envoy-split-streams/run.py b/tests/envoy-split-streams/run.py
new file mode 100644
index 00000000..a1016796
--- /dev/null
+++ b/tests/envoy-split-streams/run.py
@@ -0,0 +1,49 @@
+#!/usr/bin/env python3
+"""Prove exact split stream bytes and public marks on genuinely qualified Bash targets."""
+import argparse
+from pathlib import Path
+import subprocess
+
+import sys
+
+# Reuse the existing proof-only staging boundary; production paths stay fixed.
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'envoy-session-startup'))
+from staging import build_peer, save, sha, stage, verified_rows
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--qualification', type=Path, required=True)
+    parser.add_argument('--output', type=Path, required=True)
+    parser.add_argument('--release', choices=('5.2.15', '5.2.37', '5.3.9'))
+    args = parser.parse_args()
+    output = args.output.resolve()
+    output.mkdir(parents=True, exist_ok=False)
+    with build_peer(output, './session') as (binary, current_awsh):
+        results = []
+        for row, entry, actual in verified_rows(args.qualification, current_awsh):
+            if args.release and args.release != row['release']:
+                continue
+            iid, context = stage(output, row['release'], entry, actual, binary)
+            with (context / 'native.log').open('w') as log:
+                result = subprocess.run(
+                    ['docker', 'run', '--rm', '--network=none',
+                     '-e', 'OMEGAFLOW_NATIVE_SPLIT_STREAMS=1',
+                     '--entrypoint=/omegaflow-runtime/bin/envoy', iid,
+                     '-test.run=^TestNativeSplitStreams', '-test.v'],
+                    stdout=log, stderr=subprocess.STDOUT, timeout=45)
+            results.append({'release': row['release'], 'qualified_awsh_sha256': current_awsh,
+                            'receipt_sha256': row['receipt_sha256'], 'native_image': iid,
+                            'test_binary_sha256': sha(binary), 'returncode': result.returncode,
+                            'log_path': str(context / 'native.log'),
+                            'log_sha256': sha(context / 'native.log')})
+            save(output / 'results.json', results)
+            if result.returncode:
+                raise ValueError('actual split stream proof failed: ' + str(context / 'native.log'))
+        if not results:
+            raise ValueError('no selected qualified target was exercised')
+        print('Native split stream evidence: ' + str(output))
+
+
+if __name__ == '__main__':
+    main()
diff --git a/tests/test_native_proof_cleanup.py b/tests/test_native_proof_cleanup.py
index 88c68f0b..1cd8a319 100644
--- a/tests/test_native_proof_cleanup.py
+++ b/tests/test_native_proof_cleanup.py
@@ -10,11 +10,11 @@
 import pytest
 
 PROOFS = Path(__file__).parent / 'envoy-session-startup'
-RUNNERS = ('private', 'public', 'ordinary', 'resize', 'lifecycle', 'gate', 'controls', 'split-setup')
+RUNNERS = ('private', 'public', 'ordinary', 'resize', 'lifecycle', 'gate', 'controls', 'split-setup', 'split-streams')
 
 
 def load(name):
-    path = PROOFS.parent / 'envoy-split-setup/run.py' if name == 'split-setup' else PROOFS / (name + '.py')
+    path = PROOFS.parent / ('envoy-' + name + '/run.py') if name in ('split-setup', 'split-streams') else PROOFS / (name + '.py')
     spec = importlib.util.spec_from_file_location('proof_' + name, path)
     module = importlib.util.module_from_spec(spec)
     spec.loader.exec_module(module)