diff --git a/services/lmstudio-proxy/main.go b/services/lmstudio-proxy/main.go
index a516ef598..ec3339868 100644
--- a/services/lmstudio-proxy/main.go
+++ b/services/lmstudio-proxy/main.go
@@ -81,6 +81,7 @@ func main() {
 	tel := telemetry.NewManager("nvpair-lmstudio-proxy", telemetry.OTLPExporter)
 
 	codec := NewCodec(transport)
+	defer codec.Close() // flush accepted frames before closing the transport
 	disc := NewDiscovery()
 	proxy := NewProxy(codec, disc, effectivePort, tel)
 	defer func() {
diff --git a/services/lmstudio-proxy/transport.go b/services/lmstudio-proxy/transport.go
index b62d6278b..ecc5a269f 100644
--- a/services/lmstudio-proxy/transport.go
+++ b/services/lmstudio-proxy/transport.go
@@ -5,21 +5,13 @@ package main
 
 import (
 	"io"
-	"os"
+	"nvpair-shared/jsonrpc"
 )
 
-type stdioTransport struct {
-	io.Reader
-	io.Writer
-}
-
 func newStdioTransport() io.ReadWriteCloser {
-	return &stdioTransport{
-		Reader: os.Stdin,
-		Writer: os.Stdout,
+	transport, err := jsonrpc.NewStdioTransport()
+	if err != nil {
+		panic("cannot initialize interruptible JSON-RPC stdio")
 	}
-}
-
-func (s *stdioTransport) Close() error {
-	return nil
+	return transport
 }
diff --git a/services/lmstudio-proxy/worker_health_test.go b/services/lmstudio-proxy/worker_health_test.go
index cd660f00a..f760ee582 100644
--- a/services/lmstudio-proxy/worker_health_test.go
+++ b/services/lmstudio-proxy/worker_health_test.go
@@ -6,7 +6,6 @@ package main
 import (
 	"bytes"
 	"encoding/json"
-	"io"
 	"log/slog"
 	"net"
 	"strings"
@@ -14,7 +13,7 @@ import (
 )
 
 // TestWorkerHealth catches a missing health dispatch, incorrect correlation or
-// payload, and a response deferred until after dispatch returns.
+// payload, and a response missing from the ordered writer.
 func TestWorkerHealth(t *testing.T) {
 	for _, rawID := range []string{"7", `"health-7"`} {
 		t.Run(rawID, func(t *testing.T) {
@@ -23,6 +22,7 @@ func TestWorkerHealth(t *testing.T) {
 			id := json.RawMessage(rawID)
 			msg := &Message{JSONRPC: "2.0", ID: &id, Method: "worker/health"}
 			p.handleMessage(msg)
+			_ = p.codec.Flush(t.Context())
 			output := out.String()
 
 			var response Message
@@ -40,7 +40,7 @@ func TestWorkerHealth(t *testing.T) {
 }
 
 // TestWorkerHealthWriteError catches a failed response being silently discarded.
-// The diagnostic must be emitted inline without request or response payloads.
+// The terminal diagnostic must contain no request or response payloads.
 func TestWorkerHealthWriteError(t *testing.T) {
 	var diagnostics bytes.Buffer
 	previous := slog.Default()
@@ -59,12 +59,12 @@ func TestWorkerHealthWriteError(t *testing.T) {
 		Params: json.RawMessage(`{"private":"health-params"}`),
 	}
 	p.handleMessage(msg)
+	_ = p.codec.Close() // join asynchronous terminal diagnostic before reading its sink
 
 	output := diagnostics.String()
-	if !strings.Contains(output, "level=ERROR") ||
-		!strings.Contains(output, "worker/health") ||
-		!strings.Contains(output, io.ErrClosedPipe.Error()) {
-		t.Fatalf("health write error was not surfaced inline: %q", output)
+	if !strings.Contains(output, "JSON-RPC stream terminated") ||
+		!strings.Contains(output, `reason="write failed"`) {
+		t.Fatalf("health write error was not surfaced: %q", output)
 	}
 	for _, payload := range []string{"private-health-id", "health-params", `"ok"`} {
 		if strings.Contains(output, payload) {
diff --git a/services/nvpair-cluster-manager/main.go b/services/nvpair-cluster-manager/main.go
index 0860b3e96..52864721c 100644
--- a/services/nvpair-cluster-manager/main.go
+++ b/services/nvpair-cluster-manager/main.go
@@ -61,6 +61,7 @@ func main() {
 	}()
 
 	codec := NewCodec(transport)
+	defer codec.Close() // flush accepted frames before closing the transport
 	mgr, err := NewManager(codec, *configDir, *port)
 	if err != nil {
 		log.Fatalf("failed to initialise cluster manager: %v", err)
diff --git a/services/nvpair-cluster-manager/notification_stall_test.go b/services/nvpair-cluster-manager/notification_stall_test.go
new file mode 100644
index 000000000..79019b5d3
--- /dev/null
+++ b/services/nvpair-cluster-manager/notification_stall_test.go
@@ -0,0 +1,35 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"net"
+	"testing"
+	"time"
+)
+
+func TestPairingCompletionStalledReaderReleasesRosterMu(t *testing.T) {
+	m := newTestManager(t)
+	peer := newPinFixture(t, "peer")
+	pi, cert := pairingInfoFor(t, peer)
+	m.putInvite(&Invite{InviteID: "stalled", ClusterID: "cluster-1", State: inviteStatePending})
+	sess := putInviterSession(t, m, "stalled", "cluster-1")
+	local, remote := net.Pipe()
+	m.codec = NewCodec(local)
+	t.Cleanup(func() { _ = remote.Close(); _ = m.codec.Close() })
+	done := make(chan error, 1)
+	go func() { done <- m.finalizePairing("stalled", sess, pi, cert) }()
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatal(err)
+		}
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("pairing completion blocked under rosterMu on stalled stdout")
+	}
+	if !m.rosterMu.TryLock() {
+		t.Fatal("pairing completion retained rosterMu")
+	}
+	m.rosterMu.Unlock()
+}
diff --git a/services/nvpair-cluster-manager/pairing_publication_test.go b/services/nvpair-cluster-manager/pairing_publication_test.go
index 94f50f103..4fbdf744f 100644
--- a/services/nvpair-cluster-manager/pairing_publication_test.go
+++ b/services/nvpair-cluster-manager/pairing_publication_test.go
@@ -13,8 +13,7 @@ import (
 )
 
 type pairingPublicationWriter struct {
-	manager *Manager
-	entered chan bool
+	entered chan struct{}
 	release chan struct{}
 	mu      sync.Mutex
 	frames  []Message
@@ -27,11 +26,7 @@ func (w *pairingPublicationWriter) Write(p []byte) (int, error) {
 		return 0, err
 	}
 	if msg.Method == "cluster:pairing-completed" {
-		protected := !w.manager.rosterMu.TryLock()
-		if !protected {
-			w.manager.rosterMu.Unlock()
-		}
-		w.entered <- protected
+		w.entered <- struct{}{}
 		<-w.release
 	}
 	w.mu.Lock()
@@ -41,7 +36,7 @@ func (w *pairingPublicationWriter) Write(p []byte) (int, error) {
 }
 
 // The real completion and removal execute concurrently. The publication seam
-// observes the composition lock, and channels order the interleaving: no sleeps
+// stalls the ordered stream, and channels order the interleaving: no sleeps
 // or assumption that the broker happened to receive the completion first.
 func TestPairingCompletionPublicationOrdersConcurrentRemoval(t *testing.T) {
 	m := newTestManager(t)
@@ -51,7 +46,7 @@ func TestPairingCompletionPublicationOrdersConcurrentRemoval(t *testing.T) {
 	m.putInvite(&Invite{InviteID: "ordered", ClusterID: "cluster-1", State: inviteStatePending})
 	sess := putInviterSession(t, m, "ordered", "cluster-1")
 	sess.joinerAddr = "overlay.example:14321"
-	writer := &pairingPublicationWriter{manager: m, entered: make(chan bool, 1), release: make(chan struct{})}
+	writer := &pairingPublicationWriter{entered: make(chan struct{}, 1), release: make(chan struct{})}
 	m.codec = NewCodec(writer)
 	var once sync.Once
 	unblock := func() { once.Do(func() { close(writer.release) }) }
@@ -61,10 +56,7 @@ func TestPairingCompletionPublicationOrdersConcurrentRemoval(t *testing.T) {
 	completed := make(chan error, 1)
 	go func() { completed <- m.finalizePairing("ordered", sess, pi, cert) }()
 	select {
-	case protected := <-writer.entered:
-		if !protected {
-			t.Error("completion publication is outside the roster commit/removal boundary")
-		}
+	case <-writer.entered:
 	case <-ctx.Done():
 		t.Fatal("completion never reached publication")
 	}
@@ -91,6 +83,9 @@ func TestPairingCompletionPublicationOrdersConcurrentRemoval(t *testing.T) {
 	case <-ctx.Done():
 		t.Fatal("removal stalled")
 	}
+	if err := m.codec.Flush(ctx); err != nil {
+		t.Fatal(err)
+	}
 	writer.mu.Lock()
 	defer writer.mu.Unlock()
 	completionIndex, removalIndex := -1, -1
diff --git a/services/nvpair-cluster-manager/transport.go b/services/nvpair-cluster-manager/transport.go
index b62d6278b..ecc5a269f 100644
--- a/services/nvpair-cluster-manager/transport.go
+++ b/services/nvpair-cluster-manager/transport.go
@@ -5,21 +5,13 @@ package main
 
 import (
 	"io"
-	"os"
+	"nvpair-shared/jsonrpc"
 )
 
-type stdioTransport struct {
-	io.Reader
-	io.Writer
-}
-
 func newStdioTransport() io.ReadWriteCloser {
-	return &stdioTransport{
-		Reader: os.Stdin,
-		Writer: os.Stdout,
+	transport, err := jsonrpc.NewStdioTransport()
+	if err != nil {
+		panic("cannot initialize interruptible JSON-RPC stdio")
 	}
-}
-
-func (s *stdioTransport) Close() error {
-	return nil
+	return transport
 }
diff --git a/services/nvpair-cluster-manager/worker_health_test.go b/services/nvpair-cluster-manager/worker_health_test.go
index 071e5d965..7675275c0 100644
--- a/services/nvpair-cluster-manager/worker_health_test.go
+++ b/services/nvpair-cluster-manager/worker_health_test.go
@@ -6,7 +6,6 @@ package main
 import (
 	"bytes"
 	"encoding/json"
-	"io"
 	"log/slog"
 	"net"
 	"strings"
@@ -14,7 +13,7 @@ import (
 )
 
 // TestWorkerHealth catches a missing health dispatch, incorrect correlation or
-// payload, and a response deferred until after dispatch returns.
+// payload, and a response missing from the ordered writer.
 func TestWorkerHealth(t *testing.T) {
 	for _, rawID := range []string{"7", `"health-7"`} {
 		t.Run(rawID, func(t *testing.T) {
@@ -23,6 +22,7 @@ func TestWorkerHealth(t *testing.T) {
 			id := json.RawMessage(rawID)
 			msg := &Message{JSONRPC: "2.0", ID: &id, Method: "worker/health"}
 			m.handleMessage(msg)
+			_ = m.codec.Flush(t.Context())
 			output := out.String()
 
 			var response Message
@@ -40,7 +40,7 @@ func TestWorkerHealth(t *testing.T) {
 }
 
 // TestWorkerHealthWriteError catches a failed response being silently discarded.
-// The diagnostic must be emitted inline without request or response payloads.
+// The terminal diagnostic must contain no request or response payloads.
 func TestWorkerHealthWriteError(t *testing.T) {
 	var diagnostics bytes.Buffer
 	previous := slog.Default()
@@ -59,12 +59,12 @@ func TestWorkerHealthWriteError(t *testing.T) {
 		Params: json.RawMessage(`{"private":"health-params"}`),
 	}
 	m.handleMessage(msg)
+	_ = m.codec.Close() // join asynchronous terminal diagnostic before reading its sink
 
 	output := diagnostics.String()
-	if !strings.Contains(output, "level=ERROR") ||
-		!strings.Contains(output, "worker/health") ||
-		!strings.Contains(output, io.ErrClosedPipe.Error()) {
-		t.Fatalf("health write error was not surfaced inline: %q", output)
+	if !strings.Contains(output, "JSON-RPC stream terminated") ||
+		!strings.Contains(output, `reason="write failed"`) {
+		t.Fatalf("health write error was not surfaced: %q", output)
 	}
 	for _, payload := range []string{"private-health-id", "health-params", `"ok"`} {
 		if strings.Contains(output, payload) {
diff --git a/services/nvpair-engine-manager/main.go b/services/nvpair-engine-manager/main.go
index 907268573..3884873e7 100644
--- a/services/nvpair-engine-manager/main.go
+++ b/services/nvpair-engine-manager/main.go
@@ -89,6 +89,7 @@ func main() {
 	}()
 
 	codec := NewCodec(transport)
+	defer codec.Close() // flush accepted frames before closing the transport
 	manifestDir, installBase := userPaths()
 	reg := buildRegistry(manifestDir)
 	log.Printf("loaded %d engine manifest(s): %v", len(reg.Names()), reg.Names())
diff --git a/services/nvpair-engine-manager/transport.go b/services/nvpair-engine-manager/transport.go
index b62d6278b..ecc5a269f 100644
--- a/services/nvpair-engine-manager/transport.go
+++ b/services/nvpair-engine-manager/transport.go
@@ -5,21 +5,13 @@ package main
 
 import (
 	"io"
-	"os"
+	"nvpair-shared/jsonrpc"
 )
 
-type stdioTransport struct {
-	io.Reader
-	io.Writer
-}
-
 func newStdioTransport() io.ReadWriteCloser {
-	return &stdioTransport{
-		Reader: os.Stdin,
-		Writer: os.Stdout,
+	transport, err := jsonrpc.NewStdioTransport()
+	if err != nil {
+		panic("cannot initialize interruptible JSON-RPC stdio")
 	}
-}
-
-func (s *stdioTransport) Close() error {
-	return nil
+	return transport
 }
diff --git a/services/nvpair-engine-manager/worker_health_test.go b/services/nvpair-engine-manager/worker_health_test.go
index 0ef46757a..780ff1bc8 100644
--- a/services/nvpair-engine-manager/worker_health_test.go
+++ b/services/nvpair-engine-manager/worker_health_test.go
@@ -7,7 +7,6 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
-	"io"
 	"log/slog"
 	"net"
 	"strings"
@@ -16,7 +15,7 @@ import (
 )
 
 // TestWorkerHealth catches a missing health dispatch, incorrect correlation or
-// payload, and a response deferred until after dispatch returns.
+// payload, and a response missing from the ordered writer.
 func TestWorkerHealth(t *testing.T) {
 	for _, rawID := range []string{"7", `"health-7"`} {
 		t.Run(rawID, func(t *testing.T) {
@@ -39,6 +38,7 @@ func TestWorkerHealth(t *testing.T) {
 			done := make(chan string, 1)
 			go func() {
 				m.handleMessage(context.Background(), msg)
+				_ = m.codec.Flush(t.Context())
 				done <- out.String()
 			}()
 			var output string
@@ -63,7 +63,7 @@ func TestWorkerHealth(t *testing.T) {
 }
 
 // TestWorkerHealthWriteError catches a failed response being silently discarded.
-// The diagnostic must be emitted inline without request or response payloads.
+// The terminal diagnostic must contain no request or response payloads.
 func TestWorkerHealthWriteError(t *testing.T) {
 	var diagnostics bytes.Buffer
 	previous := slog.Default()
@@ -82,12 +82,12 @@ func TestWorkerHealthWriteError(t *testing.T) {
 		Params: json.RawMessage(`{"private":"health-params"}`),
 	}
 	m.handleMessage(context.Background(), msg)
+	_ = m.codec.Close() // join asynchronous terminal diagnostic before reading its sink
 
 	output := diagnostics.String()
-	if !strings.Contains(output, "level=ERROR") ||
-		!strings.Contains(output, "worker/health") ||
-		!strings.Contains(output, io.ErrClosedPipe.Error()) {
-		t.Fatalf("health write error was not surfaced inline: %q", output)
+	if !strings.Contains(output, "JSON-RPC stream terminated") ||
+		!strings.Contains(output, `reason="write failed"`) {
+		t.Fatalf("health write error was not surfaced: %q", output)
 	}
 	for _, payload := range []string{"private-health-id", "health-params", `"ok"`} {
 		if strings.Contains(output, payload) {
diff --git a/services/nvpair-errors/main.go b/services/nvpair-errors/main.go
index 14202698c..5a3f440b2 100644
--- a/services/nvpair-errors/main.go
+++ b/services/nvpair-errors/main.go
@@ -64,6 +64,7 @@ func main() {
 	}()
 
 	codec := NewCodec(transport)
+	defer codec.Close() // flush accepted frames before closing the transport
 	mgr := NewManager(codec)
 	// The broker passes the stable per-host UUID so this node's errors are
 	// attributed to the same identity the broker stamps and the cluster keys on
diff --git a/services/nvpair-errors/transport.go b/services/nvpair-errors/transport.go
index b62d6278b..ecc5a269f 100644
--- a/services/nvpair-errors/transport.go
+++ b/services/nvpair-errors/transport.go
@@ -5,21 +5,13 @@ package main
 
 import (
 	"io"
-	"os"
+	"nvpair-shared/jsonrpc"
 )
 
-type stdioTransport struct {
-	io.Reader
-	io.Writer
-}
-
 func newStdioTransport() io.ReadWriteCloser {
-	return &stdioTransport{
-		Reader: os.Stdin,
-		Writer: os.Stdout,
+	transport, err := jsonrpc.NewStdioTransport()
+	if err != nil {
+		panic("cannot initialize interruptible JSON-RPC stdio")
 	}
-}
-
-func (s *stdioTransport) Close() error {
-	return nil
+	return transport
 }
diff --git a/services/nvpair-errors/worker_health_test.go b/services/nvpair-errors/worker_health_test.go
index 071e5d965..7675275c0 100644
--- a/services/nvpair-errors/worker_health_test.go
+++ b/services/nvpair-errors/worker_health_test.go
@@ -6,7 +6,6 @@ package main
 import (
 	"bytes"
 	"encoding/json"
-	"io"
 	"log/slog"
 	"net"
 	"strings"
@@ -14,7 +13,7 @@ import (
 )
 
 // TestWorkerHealth catches a missing health dispatch, incorrect correlation or
-// payload, and a response deferred until after dispatch returns.
+// payload, and a response missing from the ordered writer.
 func TestWorkerHealth(t *testing.T) {
 	for _, rawID := range []string{"7", `"health-7"`} {
 		t.Run(rawID, func(t *testing.T) {
@@ -23,6 +22,7 @@ func TestWorkerHealth(t *testing.T) {
 			id := json.RawMessage(rawID)
 			msg := &Message{JSONRPC: "2.0", ID: &id, Method: "worker/health"}
 			m.handleMessage(msg)
+			_ = m.codec.Flush(t.Context())
 			output := out.String()
 
 			var response Message
@@ -40,7 +40,7 @@ func TestWorkerHealth(t *testing.T) {
 }
 
 // TestWorkerHealthWriteError catches a failed response being silently discarded.
-// The diagnostic must be emitted inline without request or response payloads.
+// The terminal diagnostic must contain no request or response payloads.
 func TestWorkerHealthWriteError(t *testing.T) {
 	var diagnostics bytes.Buffer
 	previous := slog.Default()
@@ -59,12 +59,12 @@ func TestWorkerHealthWriteError(t *testing.T) {
 		Params: json.RawMessage(`{"private":"health-params"}`),
 	}
 	m.handleMessage(msg)
+	_ = m.codec.Close() // join asynchronous terminal diagnostic before reading its sink
 
 	output := diagnostics.String()
-	if !strings.Contains(output, "level=ERROR") ||
-		!strings.Contains(output, "worker/health") ||
-		!strings.Contains(output, io.ErrClosedPipe.Error()) {
-		t.Fatalf("health write error was not surfaced inline: %q", output)
+	if !strings.Contains(output, "JSON-RPC stream terminated") ||
+		!strings.Contains(output, `reason="write failed"`) {
+		t.Fatalf("health write error was not surfaced: %q", output)
 	}
 	for _, payload := range []string{"private-health-id", "health-params", `"ok"`} {
 		if strings.Contains(output, payload) {
diff --git a/services/nvpair-job-scheduler/main.go b/services/nvpair-job-scheduler/main.go
index 357a060f1..8890e17e2 100644
--- a/services/nvpair-job-scheduler/main.go
+++ b/services/nvpair-job-scheduler/main.go
@@ -64,6 +64,7 @@ func main() {
 	}()
 
 	codec := NewCodec(transport)
+	defer codec.Close() // flush accepted frames before closing the transport
 	mgr := NewManager(codec, *interval)
 
 	if err := mgr.Run(ctx); err != nil && ctx.Err() == nil {
diff --git a/services/nvpair-job-scheduler/transport.go b/services/nvpair-job-scheduler/transport.go
index b62d6278b..ecc5a269f 100644
--- a/services/nvpair-job-scheduler/transport.go
+++ b/services/nvpair-job-scheduler/transport.go
@@ -5,21 +5,13 @@ package main
 
 import (
 	"io"
-	"os"
+	"nvpair-shared/jsonrpc"
 )
 
-type stdioTransport struct {
-	io.Reader
-	io.Writer
-}
-
 func newStdioTransport() io.ReadWriteCloser {
-	return &stdioTransport{
-		Reader: os.Stdin,
-		Writer: os.Stdout,
+	transport, err := jsonrpc.NewStdioTransport()
+	if err != nil {
+		panic("cannot initialize interruptible JSON-RPC stdio")
 	}
-}
-
-func (s *stdioTransport) Close() error {
-	return nil
+	return transport
 }
diff --git a/services/nvpair-job-scheduler/worker_health_test.go b/services/nvpair-job-scheduler/worker_health_test.go
index 071e5d965..7675275c0 100644
--- a/services/nvpair-job-scheduler/worker_health_test.go
+++ b/services/nvpair-job-scheduler/worker_health_test.go
@@ -6,7 +6,6 @@ package main
 import (
 	"bytes"
 	"encoding/json"
-	"io"
 	"log/slog"
 	"net"
 	"strings"
@@ -14,7 +13,7 @@ import (
 )
 
 // TestWorkerHealth catches a missing health dispatch, incorrect correlation or
-// payload, and a response deferred until after dispatch returns.
+// payload, and a response missing from the ordered writer.
 func TestWorkerHealth(t *testing.T) {
 	for _, rawID := range []string{"7", `"health-7"`} {
 		t.Run(rawID, func(t *testing.T) {
@@ -23,6 +22,7 @@ func TestWorkerHealth(t *testing.T) {
 			id := json.RawMessage(rawID)
 			msg := &Message{JSONRPC: "2.0", ID: &id, Method: "worker/health"}
 			m.handleMessage(msg)
+			_ = m.codec.Flush(t.Context())
 			output := out.String()
 
 			var response Message
@@ -40,7 +40,7 @@ func TestWorkerHealth(t *testing.T) {
 }
 
 // TestWorkerHealthWriteError catches a failed response being silently discarded.
-// The diagnostic must be emitted inline without request or response payloads.
+// The terminal diagnostic must contain no request or response payloads.
 func TestWorkerHealthWriteError(t *testing.T) {
 	var diagnostics bytes.Buffer
 	previous := slog.Default()
@@ -59,12 +59,12 @@ func TestWorkerHealthWriteError(t *testing.T) {
 		Params: json.RawMessage(`{"private":"health-params"}`),
 	}
 	m.handleMessage(msg)
+	_ = m.codec.Close() // join asynchronous terminal diagnostic before reading its sink
 
 	output := diagnostics.String()
-	if !strings.Contains(output, "level=ERROR") ||
-		!strings.Contains(output, "worker/health") ||
-		!strings.Contains(output, io.ErrClosedPipe.Error()) {
-		t.Fatalf("health write error was not surfaced inline: %q", output)
+	if !strings.Contains(output, "JSON-RPC stream terminated") ||
+		!strings.Contains(output, `reason="write failed"`) {
+		t.Fatalf("health write error was not surfaced: %q", output)
 	}
 	for _, payload := range []string{"private-health-id", "health-params", `"ok"`} {
 		if strings.Contains(output, payload) {
diff --git a/services/nvpair-manual-nodes/main.go b/services/nvpair-manual-nodes/main.go
index d08cc490c..c5f443f65 100644
--- a/services/nvpair-manual-nodes/main.go
+++ b/services/nvpair-manual-nodes/main.go
@@ -80,6 +80,7 @@ func main() {
 	}()
 
 	codec := NewCodec(transport)
+	defer codec.Close() // flush accepted frames before closing the transport
 	mgr, err := NewManager(codec, tlsOpts, mesh, *clusterDir)
 	if err != nil {
 		log.Fatalf("failed to construct manager: %v", err)
diff --git a/services/nvpair-manual-nodes/transport.go b/services/nvpair-manual-nodes/transport.go
index b62d6278b..ecc5a269f 100644
--- a/services/nvpair-manual-nodes/transport.go
+++ b/services/nvpair-manual-nodes/transport.go
@@ -5,21 +5,13 @@ package main
 
 import (
 	"io"
-	"os"
+	"nvpair-shared/jsonrpc"
 )
 
-type stdioTransport struct {
-	io.Reader
-	io.Writer
-}
-
 func newStdioTransport() io.ReadWriteCloser {
-	return &stdioTransport{
-		Reader: os.Stdin,
-		Writer: os.Stdout,
+	transport, err := jsonrpc.NewStdioTransport()
+	if err != nil {
+		panic("cannot initialize interruptible JSON-RPC stdio")
 	}
-}
-
-func (s *stdioTransport) Close() error {
-	return nil
+	return transport
 }
diff --git a/services/nvpair-manual-nodes/worker_health_test.go b/services/nvpair-manual-nodes/worker_health_test.go
index 071e5d965..7675275c0 100644
--- a/services/nvpair-manual-nodes/worker_health_test.go
+++ b/services/nvpair-manual-nodes/worker_health_test.go
@@ -6,7 +6,6 @@ package main
 import (
 	"bytes"
 	"encoding/json"
-	"io"
 	"log/slog"
 	"net"
 	"strings"
@@ -14,7 +13,7 @@ import (
 )
 
 // TestWorkerHealth catches a missing health dispatch, incorrect correlation or
-// payload, and a response deferred until after dispatch returns.
+// payload, and a response missing from the ordered writer.
 func TestWorkerHealth(t *testing.T) {
 	for _, rawID := range []string{"7", `"health-7"`} {
 		t.Run(rawID, func(t *testing.T) {
@@ -23,6 +22,7 @@ func TestWorkerHealth(t *testing.T) {
 			id := json.RawMessage(rawID)
 			msg := &Message{JSONRPC: "2.0", ID: &id, Method: "worker/health"}
 			m.handleMessage(msg)
+			_ = m.codec.Flush(t.Context())
 			output := out.String()
 
 			var response Message
@@ -40,7 +40,7 @@ func TestWorkerHealth(t *testing.T) {
 }
 
 // TestWorkerHealthWriteError catches a failed response being silently discarded.
-// The diagnostic must be emitted inline without request or response payloads.
+// The terminal diagnostic must contain no request or response payloads.
 func TestWorkerHealthWriteError(t *testing.T) {
 	var diagnostics bytes.Buffer
 	previous := slog.Default()
@@ -59,12 +59,12 @@ func TestWorkerHealthWriteError(t *testing.T) {
 		Params: json.RawMessage(`{"private":"health-params"}`),
 	}
 	m.handleMessage(msg)
+	_ = m.codec.Close() // join asynchronous terminal diagnostic before reading its sink
 
 	output := diagnostics.String()
-	if !strings.Contains(output, "level=ERROR") ||
-		!strings.Contains(output, "worker/health") ||
-		!strings.Contains(output, io.ErrClosedPipe.Error()) {
-		t.Fatalf("health write error was not surfaced inline: %q", output)
+	if !strings.Contains(output, "JSON-RPC stream terminated") ||
+		!strings.Contains(output, `reason="write failed"`) {
+		t.Fatalf("health write error was not surfaced: %q", output)
 	}
 	for _, payload := range []string{"private-health-id", "health-params", `"ok"`} {
 		if strings.Contains(output, payload) {
diff --git a/services/nvpair-node-scanner/main.go b/services/nvpair-node-scanner/main.go
index 9c8d9b09a..972672159 100644
--- a/services/nvpair-node-scanner/main.go
+++ b/services/nvpair-node-scanner/main.go
@@ -84,6 +84,7 @@ func main() {
 	}()
 
 	codec := NewCodec(transport)
+	defer codec.Close() // flush accepted frames before closing the transport
 	scanner := NewScanner(codec, mesh, *clusterDir, tlsHTTP)
 
 	if err := scanner.Run(ctx); err != nil && ctx.Err() == nil {
diff --git a/services/nvpair-node-scanner/transport.go b/services/nvpair-node-scanner/transport.go
index b62d6278b..ecc5a269f 100644
--- a/services/nvpair-node-scanner/transport.go
+++ b/services/nvpair-node-scanner/transport.go
@@ -5,21 +5,13 @@ package main
 
 import (
 	"io"
-	"os"
+	"nvpair-shared/jsonrpc"
 )
 
-type stdioTransport struct {
-	io.Reader
-	io.Writer
-}
-
 func newStdioTransport() io.ReadWriteCloser {
-	return &stdioTransport{
-		Reader: os.Stdin,
-		Writer: os.Stdout,
+	transport, err := jsonrpc.NewStdioTransport()
+	if err != nil {
+		panic("cannot initialize interruptible JSON-RPC stdio")
 	}
-}
-
-func (s *stdioTransport) Close() error {
-	return nil
+	return transport
 }
diff --git a/services/nvpair-node-scanner/worker_health_test.go b/services/nvpair-node-scanner/worker_health_test.go
index edb85e649..e9e23506d 100644
--- a/services/nvpair-node-scanner/worker_health_test.go
+++ b/services/nvpair-node-scanner/worker_health_test.go
@@ -6,7 +6,6 @@ package main
 import (
 	"bytes"
 	"encoding/json"
-	"io"
 	"log/slog"
 	"net"
 	"strings"
@@ -14,7 +13,7 @@ import (
 )
 
 // TestWorkerHealth catches a missing health dispatch, incorrect correlation or
-// payload, and a response deferred until after dispatch returns.
+// payload, and a response missing from the ordered writer.
 func TestWorkerHealth(t *testing.T) {
 	for _, rawID := range []string{"7", `"health-7"`} {
 		t.Run(rawID, func(t *testing.T) {
@@ -23,6 +22,7 @@ func TestWorkerHealth(t *testing.T) {
 			id := json.RawMessage(rawID)
 			msg := &Message{JSONRPC: "2.0", ID: &id, Method: "worker/health"}
 			s.handleMessage(msg)
+			_ = s.codec.Flush(t.Context())
 			output := out.String()
 
 			var response Message
@@ -40,7 +40,7 @@ func TestWorkerHealth(t *testing.T) {
 }
 
 // TestWorkerHealthWriteError catches a failed response being silently discarded.
-// The diagnostic must be emitted inline without request or response payloads.
+// The terminal diagnostic must contain no request or response payloads.
 func TestWorkerHealthWriteError(t *testing.T) {
 	var diagnostics bytes.Buffer
 	previous := slog.Default()
@@ -59,12 +59,12 @@ func TestWorkerHealthWriteError(t *testing.T) {
 		Params: json.RawMessage(`{"private":"health-params"}`),
 	}
 	s.handleMessage(msg)
+	_ = s.codec.Close() // join asynchronous terminal diagnostic before reading its sink
 
 	output := diagnostics.String()
-	if !strings.Contains(output, "level=ERROR") ||
-		!strings.Contains(output, "worker/health") ||
-		!strings.Contains(output, io.ErrClosedPipe.Error()) {
-		t.Fatalf("health write error was not surfaced inline: %q", output)
+	if !strings.Contains(output, "JSON-RPC stream terminated") ||
+		!strings.Contains(output, `reason="write failed"`) {
+		t.Fatalf("health write error was not surfaced: %q", output)
 	}
 	for _, payload := range []string{"private-health-id", "health-params", `"ok"`} {
 		if strings.Contains(output, payload) {
diff --git a/services/nvpair-node-settings/main.go b/services/nvpair-node-settings/main.go
index 10976fb12..207358300 100644
--- a/services/nvpair-node-settings/main.go
+++ b/services/nvpair-node-settings/main.go
@@ -80,6 +80,7 @@ func main() {
 	}()
 
 	codec := NewCodec(transport)
+	defer codec.Close() // flush accepted frames before closing the transport
 	mgr, err := NewManager(codec, path)
 	if err != nil {
 		log.Fatalf("failed to initialise settings manager: %v", err)
diff --git a/services/nvpair-node-settings/transport.go b/services/nvpair-node-settings/transport.go
index b62d6278b..ecc5a269f 100644
--- a/services/nvpair-node-settings/transport.go
+++ b/services/nvpair-node-settings/transport.go
@@ -5,21 +5,13 @@ package main
 
 import (
 	"io"
-	"os"
+	"nvpair-shared/jsonrpc"
 )
 
-type stdioTransport struct {
-	io.Reader
-	io.Writer
-}
-
 func newStdioTransport() io.ReadWriteCloser {
-	return &stdioTransport{
-		Reader: os.Stdin,
-		Writer: os.Stdout,
+	transport, err := jsonrpc.NewStdioTransport()
+	if err != nil {
+		panic("cannot initialize interruptible JSON-RPC stdio")
 	}
-}
-
-func (s *stdioTransport) Close() error {
-	return nil
+	return transport
 }
diff --git a/services/nvpair-node-settings/worker_health_test.go b/services/nvpair-node-settings/worker_health_test.go
index 071e5d965..7675275c0 100644
--- a/services/nvpair-node-settings/worker_health_test.go
+++ b/services/nvpair-node-settings/worker_health_test.go
@@ -6,7 +6,6 @@ package main
 import (
 	"bytes"
 	"encoding/json"
-	"io"
 	"log/slog"
 	"net"
 	"strings"
@@ -14,7 +13,7 @@ import (
 )
 
 // TestWorkerHealth catches a missing health dispatch, incorrect correlation or
-// payload, and a response deferred until after dispatch returns.
+// payload, and a response missing from the ordered writer.
 func TestWorkerHealth(t *testing.T) {
 	for _, rawID := range []string{"7", `"health-7"`} {
 		t.Run(rawID, func(t *testing.T) {
@@ -23,6 +22,7 @@ func TestWorkerHealth(t *testing.T) {
 			id := json.RawMessage(rawID)
 			msg := &Message{JSONRPC: "2.0", ID: &id, Method: "worker/health"}
 			m.handleMessage(msg)
+			_ = m.codec.Flush(t.Context())
 			output := out.String()
 
 			var response Message
@@ -40,7 +40,7 @@ func TestWorkerHealth(t *testing.T) {
 }
 
 // TestWorkerHealthWriteError catches a failed response being silently discarded.
-// The diagnostic must be emitted inline without request or response payloads.
+// The terminal diagnostic must contain no request or response payloads.
 func TestWorkerHealthWriteError(t *testing.T) {
 	var diagnostics bytes.Buffer
 	previous := slog.Default()
@@ -59,12 +59,12 @@ func TestWorkerHealthWriteError(t *testing.T) {
 		Params: json.RawMessage(`{"private":"health-params"}`),
 	}
 	m.handleMessage(msg)
+	_ = m.codec.Close() // join asynchronous terminal diagnostic before reading its sink
 
 	output := diagnostics.String()
-	if !strings.Contains(output, "level=ERROR") ||
-		!strings.Contains(output, "worker/health") ||
-		!strings.Contains(output, io.ErrClosedPipe.Error()) {
-		t.Fatalf("health write error was not surfaced inline: %q", output)
+	if !strings.Contains(output, "JSON-RPC stream terminated") ||
+		!strings.Contains(output, `reason="write failed"`) {
+		t.Fatalf("health write error was not surfaced: %q", output)
 	}
 	for _, payload := range []string{"private-health-id", "health-params", `"ok"`} {
 		if strings.Contains(output, payload) {
diff --git a/services/nvpair-tui/rpc/client.go b/services/nvpair-tui/rpc/client.go
index d37ccdb37..71de1fc7f 100644
--- a/services/nvpair-tui/rpc/client.go
+++ b/services/nvpair-tui/rpc/client.go
@@ -120,6 +120,8 @@ func (c *Client) Run(ctx context.Context) (runErr error) {
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
+		case <-c.codec.Done():
+			return c.codec.Err()
 		case frame = <-frames:
 		}
 		msg, err := frame.message, frame.err
diff --git a/services/nvpair-tui/rpc/codec.go b/services/nvpair-tui/rpc/codec.go
index 5870f7cf7..db86e5043 100644
--- a/services/nvpair-tui/rpc/codec.go
+++ b/services/nvpair-tui/rpc/codec.go
@@ -45,3 +45,12 @@ type readWriter struct {
 	io.Reader
 	io.Writer
 }
+
+// EOF on control asks the broker to stop. Its process owner must keep stdout
+// draining through worker cleanup, then close the read half after joining it.
+func (rw readWriter) Close() error {
+	if w, ok := rw.Writer.(io.Closer); ok {
+		return w.Close()
+	}
+	return nil
+}
diff --git a/services/nvpair-ui-broker/README.md b/services/nvpair-ui-broker/README.md
index f44e5bcb4..baf192d0e 100644
--- a/services/nvpair-ui-broker/README.md
+++ b/services/nvpair-ui-broker/README.md
@@ -226,6 +226,14 @@ All Go stdio readers validate through the shared codec: only malformed-frame
 DecodeError is recoverable; envelope mismatch is terminal with a metadata-only
 protocol-mismatch diagnostic. Worker reader termination closes its owned control
 pipe, so EOF reaches the child and normal supervision observes its exit.
+Outbound codec frames share one ordered writer: at most 256 outstanding frames
+and 16 MiB, with a five-second deadline from admission (including queue time).
+Notify/Respond wait at most 10 ms for immediate completion and otherwise return
+acceptance, not delivery. Overflow or a write deadline terminates the stream;
+notifications are never dropped while keeping the stream alive. Clean owners
+drain accepted frames before control EOF, within existing shutdown budgets;
+terminal failures retain the existing supervision/recovery policy. See the
+[codec sizing measurements and lifecycle audit](../shared/jsonrpc/README.md).
 Separate HTTP/mTLS cluster payloads are not stdio envelopes. Desktop TypeScript
 enforces the same canonical envelope rules as Go: duplicate or case-variant
 reserved members are recoverable decode errors, and envelope mismatches (missing
diff --git a/services/nvpair-ui-broker/broker.go b/services/nvpair-ui-broker/broker.go
index 1b4d08bbe..9afdaf5ee 100644
--- a/services/nvpair-ui-broker/broker.go
+++ b/services/nvpair-ui-broker/broker.go
@@ -3484,6 +3484,12 @@ func (b *Broker) handleMessage(msg *Message) {
 		if err := b.codec.Respond(msg.ID, nil); err != nil {
 			log.Printf("failed to respond to shutdown: %v", err)
 		}
+		// Respond now means admission, not delivery. Flush the shutdown ack
+		// before cancellation aborts this stream, using the existing ingress budget.
+		beginTeardown()
+		flushCtx, flushCancel := context.WithTimeout(context.Background(), teardownWaitReserving(250*time.Millisecond, enginePreparationReserve+workerCleanupReserve))
+		_ = b.codec.Drain(flushCtx)
+		flushCancel()
 		slog.Info("shutdown requested via JSON-RPC")
 		b.cancel()
 
diff --git a/services/nvpair-ui-broker/clustermanager.go b/services/nvpair-ui-broker/clustermanager.go
index a57188274..a988ee88d 100644
--- a/services/nvpair-ui-broker/clustermanager.go
+++ b/services/nvpair-ui-broker/clustermanager.go
@@ -179,5 +179,6 @@ func (c *clusterManagerProcess) Call(ctx context.Context, method string, params
 // EOF) and waits for it to exit, escalating if it does not — see
 // waitForStdinClose. Safe to call multiple times.
 func (c *clusterManagerProcess) Stop() {
+	flushWorkerWrites("cluster-manager", c.peer)
 	waitForStdinClose("cluster-manager", c.cmd, c.stdin, c.done)
 }
diff --git a/services/nvpair-ui-broker/codec.go b/services/nvpair-ui-broker/codec.go
index 2e0a2a7bd..87da93e09 100644
--- a/services/nvpair-ui-broker/codec.go
+++ b/services/nvpair-ui-broker/codec.go
@@ -38,6 +38,18 @@ type readWriter struct {
 	io.Writer
 }
 
+// Close interrupts both halves when the codec declares this stream dead.
+func (rw readWriter) Close() error {
+	var err error
+	if w, ok := rw.Writer.(io.Closer); ok {
+		err = w.Close()
+	}
+	if r, ok := rw.Reader.(io.Closer); ok {
+		_ = r.Close()
+	}
+	return err
+}
+
 // writeOnlyReadWriter adapts a write-only child stream — a stdin pipe — into the
 // io.ReadWriter the shared Codec expects. These handles only push control frames
 // down a child's stdin and never read it: the child's stdout is drained by the
@@ -49,3 +61,10 @@ type writeOnlyReadWriter struct {
 }
 
 func (writeOnlyReadWriter) Read([]byte) (int, error) { return 0, io.EOF }
+
+func (rw writeOnlyReadWriter) Close() error {
+	if w, ok := rw.Writer.(io.Closer); ok {
+		return w.Close()
+	}
+	return nil
+}
diff --git a/services/nvpair-ui-broker/discovery.go b/services/nvpair-ui-broker/discovery.go
index eb9b8f316..299b574a8 100644
--- a/services/nvpair-ui-broker/discovery.go
+++ b/services/nvpair-ui-broker/discovery.go
@@ -562,6 +562,7 @@ func startScanner(
 // observes as EOF) and waits for it to exit, escalating if it does not — see
 // waitForStdinClose. Safe to call multiple times.
 func (s *scannerProcess) Stop() {
+	flushWorkerWrites("scanner", s.peer)
 	waitForStdinClose("scanner", s.cmd, s.stdin, s.done)
 }
 
diff --git a/services/nvpair-ui-broker/errors.go b/services/nvpair-ui-broker/errors.go
index 9a94528e5..f157c92c5 100644
--- a/services/nvpair-ui-broker/errors.go
+++ b/services/nvpair-ui-broker/errors.go
@@ -189,5 +189,6 @@ func (e *errorsProcess) handleSubscribe(params json.RawMessage) {
 // waits for it to exit, escalating if it does not — see waitForStdinClose. Safe
 // to call multiple times.
 func (e *errorsProcess) Stop() {
+	flushWorkerWrites("errors", e.peer)
 	waitForStdinClose("errors", e.cmd, e.stdin, e.done)
 }
diff --git a/services/nvpair-ui-broker/lmstudioport_test.go b/services/nvpair-ui-broker/lmstudioport_test.go
index 32bae1697..d845438ab 100644
--- a/services/nvpair-ui-broker/lmstudioport_test.go
+++ b/services/nvpair-ui-broker/lmstudioport_test.go
@@ -11,6 +11,7 @@ import (
 	"log/slog"
 	"net"
 	"reflect"
+	"sync"
 	"testing"
 	"time"
 )
@@ -371,6 +372,10 @@ func TestManagedLMStudioConcurrentReadyWaitsForFallbackRebind(t *testing.T) {
 	b.lmstudioBackendPort.Store(managedLMStudioBackendStart)
 	b.setLMStudioProxy(proxy)
 	b.setErrors(&errorsProcess{peer: NewPeer(NewCodec(errorsClient))})
+	fallbackEntered, fallbackRelease := make(chan struct{}), make(chan struct{})
+	releaseFallback := sync.OnceFunc(func() { close(fallbackRelease) })
+	announceFallback := sync.OnceFunc(func() { close(fallbackEntered) })
+	t.Cleanup(releaseFallback)
 
 	requests := make(chan string, 4)
 	go func() {
@@ -381,12 +386,27 @@ func TestManagedLMStudioConcurrentReadyWaitsForFallbackRebind(t *testing.T) {
 			if err != nil {
 				return
 			}
+			if msg.Method != "set-port" {
+				// Reconciliation also publishes node/set-local-backend asynchronously.
+				// It is not another fallback entry and must not close the latch again.
+				_ = codec.Respond(msg.ID, nil)
+				continue
+			}
 			count++
 			requests <- msg.Method
-			if msg.Method == "set-port" && count == 1 {
+			if count == 1 {
 				_ = codec.RespondError(msg.ID, -32000, "occupied")
 			} else {
-				_ = codec.Respond(msg.ID, proxyReadyParams{Port: managedLMStudioBackendStart + 1})
+				announceFallback()
+				<-fallbackRelease
+				var requested struct {
+					Port int `json:"port"`
+				}
+				if err := json.Unmarshal(msg.Params, &requested); err != nil {
+					t.Error(err)
+					return
+				}
+				_ = codec.Respond(msg.ID, proxyReadyParams{Port: requested.Port})
 			}
 		}
 	}()
@@ -410,8 +430,13 @@ func TestManagedLMStudioConcurrentReadyWaitsForFallbackRebind(t *testing.T) {
 		}
 	}
 
-	// The first reconciliation is now blocked writing its warning. A duplicate
-	// ready must not overtake it and release the gate before the fallback move.
+	select {
+	case <-fallbackEntered:
+	case <-time.After(2 * time.Second):
+		t.Fatal("missing fallback rebind")
+	}
+	// Warning notifications no longer block reconciliation. Stall the actual
+	// fallback acknowledgement: duplicate ready must not open the gate early.
 	b.forwardLMStudioProxyNotification("ready", json.RawMessage(`{"port":1235}`))
 	select {
 	case <-b.lmstudioPortReady:
@@ -422,6 +447,7 @@ func TestManagedLMStudioConcurrentReadyWaitsForFallbackRebind(t *testing.T) {
 	if _, err := NewCodec(errorsServer).Read(); err != nil {
 		t.Fatalf("read fallback warning: %v", err)
 	}
+	releaseFallback()
 	select {
 	case <-b.lmstudioPortReady:
 	case <-time.After(2 * time.Second):
diff --git a/services/nvpair-ui-broker/main.go b/services/nvpair-ui-broker/main.go
index 344dbd62d..7db6caeeb 100644
--- a/services/nvpair-ui-broker/main.go
+++ b/services/nvpair-ui-broker/main.go
@@ -269,6 +269,7 @@ func main() {
 	}()
 
 	codec := NewCodec(transport)
+	defer codec.Close() // flush accepted frames before closing the transport
 	paths := workerPaths{
 		workerHealthInterval: *workerHealthInterval,
 		scanner:              resolvedScanner,
diff --git a/services/nvpair-ui-broker/nodeinfo.go b/services/nvpair-ui-broker/nodeinfo.go
index 62de018f6..b81ce75b5 100644
--- a/services/nvpair-ui-broker/nodeinfo.go
+++ b/services/nvpair-ui-broker/nodeinfo.go
@@ -156,14 +156,23 @@ func drainNodeInfoStdout(stdout io.Reader, onNotify func(method string, params j
 }
 
 func nodeInfoPeer(stdout io.Reader, stdin io.Writer) *Peer {
-	frames := &nodeInfoFrames{reader: bufio.NewReaderSize(stdout, maxNodeInfoLine)}
+	closer, _ := stdout.(io.Closer)
+	frames := &nodeInfoFrames{reader: bufio.NewReaderSize(stdout, maxNodeInfoLine), closer: closer}
 	return NewPeer(jsonrpc.NewCodecMaxFrame(readWriter{frames, stdin}, maxNodeInfoLine))
 }
 
 // nodeInfoFrames yields node-info's stdout to the codec one complete frame at a
 // time, dropping any frame longer than maxNodeInfoLine. Each Read returns bytes
 // from at most one frame, so the codec's buffer never holds more than one.
+func (f *nodeInfoFrames) Close() error {
+	if f.closer != nil {
+		return f.closer.Close()
+	}
+	return nil
+}
+
 type nodeInfoFrames struct {
+	closer  io.Closer
 	reader  *bufio.Reader
 	pending []byte // unread remainder of the current frame (reader's buffer)
 	err     error  // read error to report once pending is drained
@@ -204,5 +213,6 @@ func discardToFrameEnd(reader *bufio.Reader) error {
 // it to exit, escalating if it does not — see waitForStdinClose. Safe to call
 // multiple times.
 func (n *nodeInfoProcess) Stop() {
+	flushWorkerWrites("node-info", n.peer)
 	waitForStdinClose("node-info", n.cmd, n.stdin, n.done)
 }
diff --git a/services/nvpair-ui-broker/notification_burst_test.go b/services/nvpair-ui-broker/notification_burst_test.go
new file mode 100644
index 000000000..8e4dc109f
--- /dev/null
+++ b/services/nvpair-ui-broker/notification_burst_test.go
@@ -0,0 +1,97 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net"
+	"testing"
+	"time"
+
+	"nvpair-shared/jsonrpc"
+	"nvpair-shared/noderec"
+)
+
+type burstNotification struct {
+	method string
+	params json.RawMessage
+}
+
+// Synthetic startup fixture, not a claim about the largest possible cluster:
+// two lifecycle transitions per worker, 128 node deltas and a full directory.
+func notificationBurst() []burstNotification {
+	b := &Broker{}
+	b.initWorkerRegistry()
+	workers := b.workerSnapshot()
+	for i := range workers.Workers {
+		workers.Workers[i].Since = time.Unix(0, 0).UTC()
+		workers.Workers[i].LastError = ""
+	}
+	var burst []burstNotification
+	for range 24 {
+		raw, _ := json.Marshal(workers)
+		burst = append(burst, burstNotification{"workers:update", raw})
+	}
+	nodes := make([]noderec.DirectoryNode, 128)
+	for i := range nodes {
+		models := make([]string, 16)
+		for j := range models {
+			models[j] = fmt.Sprintf("fixture-model-%02d", j)
+		}
+		nodes[i] = noderec.DirectoryNode{
+			HostUUID: fmt.Sprintf("00000000-0000-0000-0000-%012d", i), Name: "fixture-node", IP: "192.0.2.1",
+			ClusterUUID: "fixture-cluster", PeerRecordID: "fixture-proof", IdentityVerification: "verified", Trusted: true,
+			Services: map[noderec.ServiceKey]noderec.ServiceStatus{noderec.ServiceCluster: {Port: 14321}, noderec.ServiceEngineManager: {Port: 14322}},
+			Models:   models,
+		}
+		raw, _ := json.Marshal(nodes[i])
+		burst = append(burst, burstNotification{noderec.NotifyNodeDiscovered, raw})
+	}
+	raw, _ := json.Marshal(noderec.GetNodesResult{Nodes: nodes})
+	return append(burst, burstNotification{noderec.NotifyNodes, raw})
+}
+
+func measureNotificationBurst(tb testing.TB, burst []burstNotification) (int, int, time.Duration) {
+	tb.Helper()
+	local, remote := net.Pipe()
+	c := NewCodec(local)
+	defer c.Close()
+	defer remote.Close()
+	received := make(chan []byte, 1)
+	go func() { var buf bytes.Buffer; _, _ = io.Copy(&buf, remote); received <- buf.Bytes() }()
+	start := time.Now()
+	for _, frame := range burst {
+		if err := c.Notify(frame.method, frame.params); err != nil {
+			tb.Fatal(err)
+		}
+	}
+	if err := c.Flush(context.Background()); err != nil {
+		tb.Fatal(err)
+	}
+	elapsed := time.Since(start)
+	_ = local.Close()
+	wire := <-received
+	return bytes.Count(wire, []byte{'\n'}), len(wire), elapsed
+}
+
+func TestNotificationBurstBudget(t *testing.T) {
+	frames, size, elapsed := measureNotificationBurst(t, notificationBurst())
+	t.Logf("startup fixture: %d frames, %d bytes, drain %s", frames, size, elapsed)
+	if frames != 153 || frames >= jsonrpc.WriteQueueFrames || size >= jsonrpc.WriteQueueBytes {
+		t.Fatalf("fixture no longer fits admission budget: %d frames, %d bytes", frames, size)
+	}
+}
+
+func BenchmarkNotificationBurst(b *testing.B) {
+	burst := notificationBurst()
+	b.ResetTimer()
+	for range b.N {
+		_, size, _ := measureNotificationBurst(b, burst)
+		b.SetBytes(int64(size))
+	}
+}
diff --git a/services/nvpair-ui-broker/notification_stall_test.go b/services/nvpair-ui-broker/notification_stall_test.go
new file mode 100644
index 000000000..ec81cf803
--- /dev/null
+++ b/services/nvpair-ui-broker/notification_stall_test.go
@@ -0,0 +1,79 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"encoding/json"
+	"net"
+	"testing"
+	"time"
+
+	serviceerrors "nvpair-shared/errors"
+	"nvpair-ui-broker/relay"
+)
+
+func stalledNotificationCodec(t *testing.T) *Codec {
+	t.Helper()
+	local, remote := net.Pipe()
+	c := NewCodec(local)
+	t.Cleanup(func() { _ = remote.Close(); _ = c.Close() })
+	return c
+}
+
+func returnsDespiteStalledReader(t *testing.T, f func()) {
+	t.Helper()
+	done := make(chan struct{})
+	go func() { f(); close(done) }()
+	select {
+	case <-done:
+	case <-time.After(200 * time.Millisecond):
+		t.Fatal("stalled notification retained caller or its lock")
+	}
+}
+
+func TestPeerMetadataStalledReaderReleasesStdinMu(t *testing.T) {
+	np := &nodeInfoProcess{codec: stalledNotificationCodec(t)}
+	b := &Broker{regCache: relay.NewRegistrationCache(), nodeID: "fixture"}
+	b.setNodeInfo(np)
+	returnsDespiteStalledReader(t, b.pushPeerMetadataToNodeInfo)
+	if !np.stdinMu.TryLock() {
+		t.Fatal("peer metadata retained stdinMu")
+	}
+	np.stdinMu.Unlock()
+}
+
+func TestErrorReportStalledReaderReturns(t *testing.T) {
+	b := &Broker{}
+	b.setErrors(&errorsProcess{peer: NewPeer(stalledNotificationCodec(t))})
+	returnsDespiteStalledReader(t, func() { b.forwardErrorsReport(serviceerrors.ServiceError{ID: "fixture"}) })
+}
+
+func TestErrorClearStalledReaderReturns(t *testing.T) {
+	b := &Broker{}
+	b.setErrors(&errorsProcess{peer: NewPeer(stalledNotificationCodec(t))})
+	returnsDespiteStalledReader(t, func() { b.forwardErrorsClear("fixture") })
+}
+
+func TestErrorReplayStalledReaderReleasesErrorMu(t *testing.T) {
+	b := &Broker{ollamaHostAliasError: &serviceerrors.ServiceError{ID: "fixture"}}
+	b.setErrors(&errorsProcess{peer: NewPeer(stalledNotificationCodec(t))})
+	returnsDespiteStalledReader(t, b.replayOllamaHostAliasError)
+	if !b.ollamaHostAliasErrorMu.TryLock() {
+		t.Fatal("error replay retained error lock")
+	}
+	b.ollamaHostAliasErrorMu.Unlock()
+}
+
+func TestPublicPortAssignmentStalledReaderReturns(t *testing.T) {
+	b := &Broker{codec: stalledNotificationCodec(t)}
+	b.managedOllamaFacade.Store(true)
+	id := json.RawMessage("1")
+	// Exercise the actual public port-assignment rejection, not Codec directly.
+	returnsDespiteStalledReader(t, func() {
+		b.relayToEngineNow(&Message{
+			JSONRPC: "2.0", ID: &id, Method: "engine:set-port",
+			Params: json.RawMessage(`{"engine":"ollama","port":11434}`),
+		})
+	})
+}
diff --git a/services/nvpair-ui-broker/proxy.go b/services/nvpair-ui-broker/proxy.go
index 09511fceb..084240bfa 100644
--- a/services/nvpair-ui-broker/proxy.go
+++ b/services/nvpair-ui-broker/proxy.go
@@ -265,5 +265,6 @@ func (p *proxyProcess) Call(ctx context.Context, method string, params json.RawM
 // proxy drains its HTTP server) and waits for it to exit, escalating if it does
 // not — see waitForStdinClose. Safe to call multiple times.
 func (p *proxyProcess) Stop() {
+	flushWorkerWrites(p.name, p.peer)
 	waitForStdinClose(p.name, p.cmd, p.stdin, p.done)
 }
diff --git a/services/nvpair-ui-broker/resync_test.go b/services/nvpair-ui-broker/resync_test.go
index ba7d99705..591d19e2b 100644
--- a/services/nvpair-ui-broker/resync_test.go
+++ b/services/nvpair-ui-broker/resync_test.go
@@ -201,10 +201,9 @@ func joinHeartbeatForCleanup(t *testing.T, stopHeartbeat func(), logs *lockedBuf
 	}
 }
 
-// TestHeartbeatStopBoundsNoopTransportClose covers stdio-shaped shutdown: Close
-// is a no-op, so a write blocked below the codec cannot be unblocked. stop must
-// abandon the join rather than park Serve forever, and it must log that single
-// abandonment without including any payload.
+// Even a defective custom transport with a no-op Close cannot retain the
+// heartbeat producer: Notify returns on bounded admission. Unlike production
+// transports, this test must explicitly release its underlying write.
 func TestHeartbeatStopBoundsNoopTransportClose(t *testing.T) {
 	writer := newNoopCloseBlockingReadWriter()
 	b := &Broker{
@@ -255,8 +254,8 @@ func TestHeartbeatStopBoundsNoopTransportClose(t *testing.T) {
 		t.Fatalf("stopHeartbeat returned after %v, want within the bounded wait", elapsed)
 	}
 	got := string(logs.Bytes())
-	if !strings.Contains(got, "heartbeat") || !strings.Contains(got, "250ms") {
-		t.Fatalf("heartbeat shutdown warning = %q, want method and timeout", got)
+	if strings.Contains(got, "abandon") {
+		t.Fatalf("bounded notification admission should let the heartbeat join: %q", got)
 	}
 }
 
diff --git a/services/nvpair-ui-broker/rpcworker.go b/services/nvpair-ui-broker/rpcworker.go
index 8e1f48f2f..1877da91f 100644
--- a/services/nvpair-ui-broker/rpcworker.go
+++ b/services/nvpair-ui-broker/rpcworker.go
@@ -143,5 +143,6 @@ func (w *rpcWorker) handleNotify(method string, params json.RawMessage) {
 // waits for it to exit, escalating if it does not — see waitForStdinClose. Safe
 // to call multiple times.
 func (w *rpcWorker) Stop() {
+	flushWorkerWrites(w.name, w.peer)
 	waitForStdinClose(w.name, w.cmd, w.stdin, w.done)
 }
diff --git a/services/nvpair-ui-broker/stopwait.go b/services/nvpair-ui-broker/stopwait.go
index 55509738f..0c71d97df 100644
--- a/services/nvpair-ui-broker/stopwait.go
+++ b/services/nvpair-ui-broker/stopwait.go
@@ -4,6 +4,7 @@
 package main
 
 import (
+	"context"
 	"io"
 	"log/slog"
 	"os"
@@ -155,6 +156,17 @@ func engineShutdownDeadline() time.Duration {
 	return teardownWaitReserving(engineStopAllBudget, workerCleanupReserve)
 }
 
+// Flush before EOF without closing the read half: workers still need stdout
+// during cleanup. This wait draws on the existing phase/teardown budget.
+func flushWorkerWrites(name string, peer *Peer) {
+	if peer == nil {
+		return
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), graceFor(name))
+	defer cancel()
+	_ = peer.Codec().Drain(ctx)
+}
+
 // waitForStdinClose signals a managed worker to shut down by closing its stdin
 // (which every worker observes as EOF and treats as its shutdown signal), then
 // waits for the process to actually exit (done is closed by the cmd.Wait
diff --git a/services/nvpair-ui-broker/transport.go b/services/nvpair-ui-broker/transport.go
index b62d6278b..ecc5a269f 100644
--- a/services/nvpair-ui-broker/transport.go
+++ b/services/nvpair-ui-broker/transport.go
@@ -5,21 +5,13 @@ package main
 
 import (
 	"io"
-	"os"
+	"nvpair-shared/jsonrpc"
 )
 
-type stdioTransport struct {
-	io.Reader
-	io.Writer
-}
-
 func newStdioTransport() io.ReadWriteCloser {
-	return &stdioTransport{
-		Reader: os.Stdin,
-		Writer: os.Stdout,
+	transport, err := jsonrpc.NewStdioTransport()
+	if err != nil {
+		panic("cannot initialize interruptible JSON-RPC stdio")
 	}
-}
-
-func (s *stdioTransport) Close() error {
-	return nil
+	return transport
 }
diff --git a/services/nvpair-ui-broker/workloadmanager.go b/services/nvpair-ui-broker/workloadmanager.go
index 75ec3e3ae..bc50fb6ca 100644
--- a/services/nvpair-ui-broker/workloadmanager.go
+++ b/services/nvpair-ui-broker/workloadmanager.go
@@ -194,5 +194,6 @@ func (m *workloadManagerProcess) handleSubscribe(params json.RawMessage) {
 // to exit, escalating if it does not — see waitForStdinClose. Safe to call
 // multiple times.
 func (m *workloadManagerProcess) Stop() {
+	flushWorkerWrites("workload-manager", m.peer)
 	waitForStdinClose("workload-manager", m.cmd, m.stdin, m.done)
 }
diff --git a/services/nvpair-workload-manager/main.go b/services/nvpair-workload-manager/main.go
index 24712bf75..3fa6123bb 100644
--- a/services/nvpair-workload-manager/main.go
+++ b/services/nvpair-workload-manager/main.go
@@ -83,6 +83,7 @@ func main() {
 	}()
 
 	codec := NewCodec(transport)
+	defer codec.Close() // flush accepted frames before closing the transport
 	mgr := NewManager(codec, *port, selfUUID, *clusterDir)
 
 	if err := mgr.Run(ctx); err != nil && ctx.Err() == nil {
diff --git a/services/nvpair-workload-manager/transport.go b/services/nvpair-workload-manager/transport.go
index b62d6278b..ecc5a269f 100644
--- a/services/nvpair-workload-manager/transport.go
+++ b/services/nvpair-workload-manager/transport.go
@@ -5,21 +5,13 @@ package main
 
 import (
 	"io"
-	"os"
+	"nvpair-shared/jsonrpc"
 )
 
-type stdioTransport struct {
-	io.Reader
-	io.Writer
-}
-
 func newStdioTransport() io.ReadWriteCloser {
-	return &stdioTransport{
-		Reader: os.Stdin,
-		Writer: os.Stdout,
+	transport, err := jsonrpc.NewStdioTransport()
+	if err != nil {
+		panic("cannot initialize interruptible JSON-RPC stdio")
 	}
-}
-
-func (s *stdioTransport) Close() error {
-	return nil
+	return transport
 }
diff --git a/services/nvpair-workload-manager/worker_health_test.go b/services/nvpair-workload-manager/worker_health_test.go
index 071e5d965..7675275c0 100644
--- a/services/nvpair-workload-manager/worker_health_test.go
+++ b/services/nvpair-workload-manager/worker_health_test.go
@@ -6,7 +6,6 @@ package main
 import (
 	"bytes"
 	"encoding/json"
-	"io"
 	"log/slog"
 	"net"
 	"strings"
@@ -14,7 +13,7 @@ import (
 )
 
 // TestWorkerHealth catches a missing health dispatch, incorrect correlation or
-// payload, and a response deferred until after dispatch returns.
+// payload, and a response missing from the ordered writer.
 func TestWorkerHealth(t *testing.T) {
 	for _, rawID := range []string{"7", `"health-7"`} {
 		t.Run(rawID, func(t *testing.T) {
@@ -23,6 +22,7 @@ func TestWorkerHealth(t *testing.T) {
 			id := json.RawMessage(rawID)
 			msg := &Message{JSONRPC: "2.0", ID: &id, Method: "worker/health"}
 			m.handleMessage(msg)
+			_ = m.codec.Flush(t.Context())
 			output := out.String()
 
 			var response Message
@@ -40,7 +40,7 @@ func TestWorkerHealth(t *testing.T) {
 }
 
 // TestWorkerHealthWriteError catches a failed response being silently discarded.
-// The diagnostic must be emitted inline without request or response payloads.
+// The terminal diagnostic must contain no request or response payloads.
 func TestWorkerHealthWriteError(t *testing.T) {
 	var diagnostics bytes.Buffer
 	previous := slog.Default()
@@ -59,12 +59,12 @@ func TestWorkerHealthWriteError(t *testing.T) {
 		Params: json.RawMessage(`{"private":"health-params"}`),
 	}
 	m.handleMessage(msg)
+	_ = m.codec.Close() // join asynchronous terminal diagnostic before reading its sink
 
 	output := diagnostics.String()
-	if !strings.Contains(output, "level=ERROR") ||
-		!strings.Contains(output, "worker/health") ||
-		!strings.Contains(output, io.ErrClosedPipe.Error()) {
-		t.Fatalf("health write error was not surfaced inline: %q", output)
+	if !strings.Contains(output, "JSON-RPC stream terminated") ||
+		!strings.Contains(output, `reason="write failed"`) {
+		t.Fatalf("health write error was not surfaced: %q", output)
 	}
 	for _, payload := range []string{"private-health-id", "health-params", `"ok"`} {
 		if strings.Contains(output, payload) {
diff --git a/services/ollama-proxy/main.go b/services/ollama-proxy/main.go
index d4c4ad163..3bfb9787f 100644
--- a/services/ollama-proxy/main.go
+++ b/services/ollama-proxy/main.go
@@ -92,6 +92,7 @@ func main() {
 	tel := telemetry.NewManager("nvpair-ollama-proxy", telemetry.OTLPExporter)
 
 	codec := NewCodec(transport)
+	defer codec.Close() // flush accepted frames before closing the transport
 	disc := NewDiscovery()
 	proxy := NewProxy(codec, disc, effectivePort, tel)
 	defer func() {
diff --git a/services/ollama-proxy/transport.go b/services/ollama-proxy/transport.go
index b62d6278b..ecc5a269f 100644
--- a/services/ollama-proxy/transport.go
+++ b/services/ollama-proxy/transport.go
@@ -5,21 +5,13 @@ package main
 
 import (
 	"io"
-	"os"
+	"nvpair-shared/jsonrpc"
 )
 
-type stdioTransport struct {
-	io.Reader
-	io.Writer
-}
-
 func newStdioTransport() io.ReadWriteCloser {
-	return &stdioTransport{
-		Reader: os.Stdin,
-		Writer: os.Stdout,
+	transport, err := jsonrpc.NewStdioTransport()
+	if err != nil {
+		panic("cannot initialize interruptible JSON-RPC stdio")
 	}
-}
-
-func (s *stdioTransport) Close() error {
-	return nil
+	return transport
 }
diff --git a/services/ollama-proxy/worker_health_test.go b/services/ollama-proxy/worker_health_test.go
index cd660f00a..f760ee582 100644
--- a/services/ollama-proxy/worker_health_test.go
+++ b/services/ollama-proxy/worker_health_test.go
@@ -6,7 +6,6 @@ package main
 import (
 	"bytes"
 	"encoding/json"
-	"io"
 	"log/slog"
 	"net"
 	"strings"
@@ -14,7 +13,7 @@ import (
 )
 
 // TestWorkerHealth catches a missing health dispatch, incorrect correlation or
-// payload, and a response deferred until after dispatch returns.
+// payload, and a response missing from the ordered writer.
 func TestWorkerHealth(t *testing.T) {
 	for _, rawID := range []string{"7", `"health-7"`} {
 		t.Run(rawID, func(t *testing.T) {
@@ -23,6 +22,7 @@ func TestWorkerHealth(t *testing.T) {
 			id := json.RawMessage(rawID)
 			msg := &Message{JSONRPC: "2.0", ID: &id, Method: "worker/health"}
 			p.handleMessage(msg)
+			_ = p.codec.Flush(t.Context())
 			output := out.String()
 
 			var response Message
@@ -40,7 +40,7 @@ func TestWorkerHealth(t *testing.T) {
 }
 
 // TestWorkerHealthWriteError catches a failed response being silently discarded.
-// The diagnostic must be emitted inline without request or response payloads.
+// The terminal diagnostic must contain no request or response payloads.
 func TestWorkerHealthWriteError(t *testing.T) {
 	var diagnostics bytes.Buffer
 	previous := slog.Default()
@@ -59,12 +59,12 @@ func TestWorkerHealthWriteError(t *testing.T) {
 		Params: json.RawMessage(`{"private":"health-params"}`),
 	}
 	p.handleMessage(msg)
+	_ = p.codec.Close() // join asynchronous terminal diagnostic before reading its sink
 
 	output := diagnostics.String()
-	if !strings.Contains(output, "level=ERROR") ||
-		!strings.Contains(output, "worker/health") ||
-		!strings.Contains(output, io.ErrClosedPipe.Error()) {
-		t.Fatalf("health write error was not surfaced inline: %q", output)
+	if !strings.Contains(output, "JSON-RPC stream terminated") ||
+		!strings.Contains(output, `reason="write failed"`) {
+		t.Fatalf("health write error was not surfaced: %q", output)
 	}
 	for _, payload := range []string{"private-health-id", "health-params", `"ok"`} {
 		if strings.Contains(output, payload) {
diff --git a/services/omlx-proxy/main.go b/services/omlx-proxy/main.go
index 672844fe7..149722a94 100644
--- a/services/omlx-proxy/main.go
+++ b/services/omlx-proxy/main.go
@@ -81,6 +81,7 @@ func main() {
 	tel := telemetry.NewManager("nvpair-omlx-proxy", telemetry.OTLPExporter)
 
 	codec := NewCodec(transport)
+	defer codec.Close() // flush accepted frames before closing the transport
 	disc := NewDiscovery()
 	proxy := NewProxy(codec, disc, effectivePort, tel)
 	defer func() {
diff --git a/services/omlx-proxy/transport.go b/services/omlx-proxy/transport.go
index b62d6278b..ecc5a269f 100644
--- a/services/omlx-proxy/transport.go
+++ b/services/omlx-proxy/transport.go
@@ -5,21 +5,13 @@ package main
 
 import (
 	"io"
-	"os"
+	"nvpair-shared/jsonrpc"
 )
 
-type stdioTransport struct {
-	io.Reader
-	io.Writer
-}
-
 func newStdioTransport() io.ReadWriteCloser {
-	return &stdioTransport{
-		Reader: os.Stdin,
-		Writer: os.Stdout,
+	transport, err := jsonrpc.NewStdioTransport()
+	if err != nil {
+		panic("cannot initialize interruptible JSON-RPC stdio")
 	}
-}
-
-func (s *stdioTransport) Close() error {
-	return nil
+	return transport
 }
diff --git a/services/omlx-proxy/worker_health_test.go b/services/omlx-proxy/worker_health_test.go
index cd660f00a..f760ee582 100644
--- a/services/omlx-proxy/worker_health_test.go
+++ b/services/omlx-proxy/worker_health_test.go
@@ -6,7 +6,6 @@ package main
 import (
 	"bytes"
 	"encoding/json"
-	"io"
 	"log/slog"
 	"net"
 	"strings"
@@ -14,7 +13,7 @@ import (
 )
 
 // TestWorkerHealth catches a missing health dispatch, incorrect correlation or
-// payload, and a response deferred until after dispatch returns.
+// payload, and a response missing from the ordered writer.
 func TestWorkerHealth(t *testing.T) {
 	for _, rawID := range []string{"7", `"health-7"`} {
 		t.Run(rawID, func(t *testing.T) {
@@ -23,6 +22,7 @@ func TestWorkerHealth(t *testing.T) {
 			id := json.RawMessage(rawID)
 			msg := &Message{JSONRPC: "2.0", ID: &id, Method: "worker/health"}
 			p.handleMessage(msg)
+			_ = p.codec.Flush(t.Context())
 			output := out.String()
 
 			var response Message
@@ -40,7 +40,7 @@ func TestWorkerHealth(t *testing.T) {
 }
 
 // TestWorkerHealthWriteError catches a failed response being silently discarded.
-// The diagnostic must be emitted inline without request or response payloads.
+// The terminal diagnostic must contain no request or response payloads.
 func TestWorkerHealthWriteError(t *testing.T) {
 	var diagnostics bytes.Buffer
 	previous := slog.Default()
@@ -59,12 +59,12 @@ func TestWorkerHealthWriteError(t *testing.T) {
 		Params: json.RawMessage(`{"private":"health-params"}`),
 	}
 	p.handleMessage(msg)
+	_ = p.codec.Close() // join asynchronous terminal diagnostic before reading its sink
 
 	output := diagnostics.String()
-	if !strings.Contains(output, "level=ERROR") ||
-		!strings.Contains(output, "worker/health") ||
-		!strings.Contains(output, io.ErrClosedPipe.Error()) {
-		t.Fatalf("health write error was not surfaced inline: %q", output)
+	if !strings.Contains(output, "JSON-RPC stream terminated") ||
+		!strings.Contains(output, `reason="write failed"`) {
+		t.Fatalf("health write error was not surfaced: %q", output)
 	}
 	for _, payload := range []string{"private-health-id", "health-params", `"ok"`} {
 		if strings.Contains(output, payload) {
diff --git a/services/shared/go.mod b/services/shared/go.mod
index e6a7f4f2f..f2eea1727 100644
--- a/services/shared/go.mod
+++ b/services/shared/go.mod
@@ -11,6 +11,7 @@ require (
 	go.opentelemetry.io/otel/sdk v1.34.0
 	go.opentelemetry.io/otel/trace v1.34.0
 	go.opentelemetry.io/proto/otlp v1.5.0
+	go.uber.org/goleak v1.3.0
 	golang.org/x/net v0.58.0
 	golang.org/x/sys v0.47.0
 	google.golang.org/grpc v1.69.4
diff --git a/services/shared/jsonrpc/README.md b/services/shared/jsonrpc/README.md
new file mode 100644
index 000000000..de671cb97
--- /dev/null
+++ b/services/shared/jsonrpc/README.md
@@ -0,0 +1,108 @@
+<!--
+SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+SPDX-License-Identifier: Apache-2.0
+-->
+
+# Ordered, bounded JSON-RPC output
+
+Each Codec has one ordered writer, started on demand and retired when idle.
+Notification stamping, serialization and queue admission share one short critical
+section; no pipe I/O or logging runs under that lock. Notifications, requests and
+responses use the **same FIFO**. Notifications alone carry epoch/seq. An accepted
+frame is immutable even if its caller later changes its buffers.
+
+## Bounds and terminal failures
+
+Admission allows **256 outstanding frames and 16 MiB**, including the active
+frame. Each accepted frame has a **5 second deadline measured from admission**,
+including time queued, not 5 seconds per partial Write. The writer handles short
+writes without permitting interleaving. Notify and response methods wait at most
+**10 ms** for the common immediate-completion case, then return acceptance, not
+delivery. Encoding and enqueue-lock CPU time are additional; producers never
+wait for pipe progress beyond that short completion window. Requests wait for
+write completion, cancellation, or the same bounded stream deadline. A queued
+request canceled before starting is skipped; an active partial frame is never
+interrupted except by killing the stream.
+
+Exhausting either queue limit, a write error, or a write deadline atomically
+marks the stream dead: all later sends fail, Done closes, and the owned
+transport is closed. Pending peer calls fail. There is **no drop-and-continue**
+mode and no automatic reconnect here; existing process owners/supervisors own
+recovery. One asynchronous terminal cleanup logs only a fixed failure category,
+never the frame, method, error string, or payload. Neither a transport Close nor
+a slow log sink can hold a producer's internal locks.
+
+### Reproducible sizing fixture
+
+Run from services/nvpair-ui-broker:
+
+```sh
+go test -run TestNotificationBurstBudget -bench '^BenchmarkNotificationBurst$' -benchtime=10x -count=1 -v .
+```
+
+The synthetic fixture uses the actual WorkersResult and DirectoryNode types:
+24 full twelve-worker lifecycle snapshots (two transitions per worker), 128 node
+deltas (16 model names each), then one full discovery directory. On macOS arm64,
+Apple M5 Max, Go 1.27.3, it produced **153 frames / 198,511 bytes**; a healthy
+net.Pipe reader drained the test burst in 2.871 ms, and ten benchmark iterations
+averaged 1.474 ms/burst. These are local synthetic measurements, not a bound on
+real cluster size or a performance guarantee. 256 frames leave 103 slots beyond
+this fixture; 16 MiB also accommodates a large engine-manager frame (its inbound
+limit is 8 MiB) plus other traffic. Five seconds leaves scheduling/large-frame
+headroom without an indefinite lock or request-handler wait. Oversized or
+sustained bursts are intentionally terminal, not unbounded memory growth.
+
+## Lifecycle and transport requirements
+
+- Flush(ctx) is a bounded barrier for already accepted output. A timeout or
+  context cancellation makes the stream dead; it is not a recoverable flush
+  error.
+- Drain(ctx) first stops admission, then flushes without closing the read half.
+  Use it before control-pipe EOF while a process owner continues draining child
+  stdout through cleanup.
+- Close stops admission and flushes within one total five-second window, then
+  closes the stream. It does not multiply the timeout by queue length.
+- BindWriteContext is **abortive** on cancellation. Call Drain/Close first for
+  clean shutdown; canceling a context is not a delivery acknowledgement.
+- Peer.Close is abortive and wakes pending calls. Clean owners drain before
+  closing control; terminal read errors and process exits abort instead.
+- Production transports must provide Close that promptly interrupts Write.
+  A permanently blocked arbitrary io.Writer with no effective Close cannot be
+  made interruptible by Go. Finite in-memory writers are supported for tests,
+  but callers inspecting their buffers must Flush first.
+- NewStdioTransport takes ownership of stdin/stdout. On Unix it transfers their
+  descriptors into nonblocking, runtime-poller registered files; merely closing
+  an inherited blocking os.Stdout can itself wait behind a blocked Write.
+  Windows uses os.File's pipe-I/O cancellation. IPC connections and subprocess
+  adapters also expose interruptible closure. Windows behavior is compile/vet
+  checked locally, not runtime-certified by Unix tests.
+
+## Codec-user audit
+
+All twelve codec-based executable entry points defer Codec.Close ahead of their
+transport close. Their synchronous Notify/Respond results now mean acceptance;
+no state machine is allowed to depend on physical delivery. Pairing publication
+under rosterMu, peer metadata under stdinMu, error report/clear/replay, and the
+public port-assignment response need no production lock restructuring: FIFO
+admission preserves their transaction ordering without pipe waits. Regression
+tests drive those real callers with an unread net.Pipe.
+
+The broker's seven worker-handle kinds drain before control EOF, clipped to the
+existing teardown/phase budget, and keep stdout available during successful
+cleanup. Their read/write adapters close both halves on terminal failure.
+The node-info frame-filter adapter forwards Close. The broker drains the
+shutdown acknowledgement inside its ingress budget before canceling its codec
+context. Worker main defers flush queued final responses/notifications.
+
+The TUI's codec closes control on failure but preserves the existing stdout-drain
+through the broker's bounded process join; its session observes Codec.Done and
+fails pending calls immediately. Its shutdown RPC waits for the response before
+control EOF. The settings projection sender remains the ordered, coalescing
+domain producer; it feeds this FIFO without relying on Notify completion.
+Tests that previously stalled that sender or other state machines by blocking a
+notification now synchronize the actual RPC acknowledgement or explicitly flush.
+
+node-info's applog StdinRPC uses Codec only as a reader; its separate Notifier
+and direct response writer are unchanged (not Codec write users). Shared
+proxyport uses Peer; its cancellation and partial-frame tests remain exercised.
+The isolated cross-process tests are consumers too, but are run only in CI.
diff --git a/services/shared/jsonrpc/envelope.go b/services/shared/jsonrpc/envelope.go
index 653eadaf6..ba5f7500c 100644
--- a/services/shared/jsonrpc/envelope.go
+++ b/services/shared/jsonrpc/envelope.go
@@ -51,8 +51,7 @@ func (e *ProtocolMismatchError) Error() string { return "protocol mismatch: " +
 // stamp assigns this connection's next seq and the process epoch to a
 // notification frame. Callers never do this themselves: Notify stamps every
 // notification it writes, so no call site has to know the envelope exists.
-// stamp is called with wmu held so that the seq it assigns is the seq that
-// reaches the wire next; seq is atomic purely for the race detector.
+// stamp runs under the queue admission lock, so seq order is FIFO wire order.
 func (c *Codec) stamp(msg *Message) {
 	epoch := Epoch()
 	seq := c.seq.Add(1)
diff --git a/services/shared/jsonrpc/envelope_test.go b/services/shared/jsonrpc/envelope_test.go
index ce22c8fb2..83b211721 100644
--- a/services/shared/jsonrpc/envelope_test.go
+++ b/services/shared/jsonrpc/envelope_test.go
@@ -24,6 +24,7 @@ func TestNotifyStampsEpochAndSeq(t *testing.T) {
 	if err := c.Notify("node/updated", map[string]string{"id": "n2"}); err != nil {
 		t.Fatalf("Notify: %v", err)
 	}
+	mustFlush(t, c)
 	lines := strings.Split(strings.TrimRight(buf.String(), "\n"), "\n")
 	if len(lines) != 2 {
 		t.Fatalf("expected 2 frames, got %d", len(lines))
@@ -60,6 +61,8 @@ func TestSeqIsPerCodecNotGlobal(t *testing.T) {
 	if err := bc.Notify("m", nil); err != nil {
 		t.Fatalf("bc.Notify: %v", err)
 	}
+	mustFlush(t, ac)
+	mustFlush(t, bc)
 	// Each codec numbers its own connection from 1, so the second codec's
 	// single frame must carry seq 1 even though the first is at seq 3.
 	if want := `"seq":1`; !strings.Contains(b.String(), want) {
@@ -74,6 +77,7 @@ func TestStampingOmitsResponsesAndRequests(t *testing.T) {
 	if err := c.Respond(&id, map[string]bool{"ok": true}); err != nil {
 		t.Fatalf("Respond: %v", err)
 	}
+	mustFlush(t, c)
 	out := buf.String()
 	if strings.Contains(out, `"epoch"`) || strings.Contains(out, `"seq"`) {
 		t.Errorf("response frame must not carry epoch/seq: %s", out)
@@ -151,6 +155,7 @@ func TestConcurrentNotifyWritesInSeqOrder(t *testing.T) {
 	}
 	close(start)
 	wg.Wait()
+	mustFlush(t, c)
 
 	// Decode every frame off the wire in order and assert wire order == seq order.
 	reader := readCodec(bytes.NewReader(buf.Bytes()))
@@ -178,7 +183,7 @@ func TestConcurrentNotifyWritesInSeqOrder(t *testing.T) {
 }
 
 // syncBuffer is an io.Writer safe to hand to a Codec that concurrent
-// goroutines notify through. The codec serialises its own writes with wmu, but
+// goroutines notify through. The codec serialises its own writes, but
 // the test does not assume that and the buffer is cheap to make safe.
 type syncBuffer struct {
 	mu  sync.Mutex
diff --git a/services/shared/jsonrpc/jsonrpc.go b/services/shared/jsonrpc/jsonrpc.go
index c4fb18371..0d859e076 100644
--- a/services/shared/jsonrpc/jsonrpc.go
+++ b/services/shared/jsonrpc/jsonrpc.go
@@ -4,13 +4,10 @@
 // Package jsonrpc is the shared newline-delimited JSON-RPC 2.0 codec used by
 // every NVPAIR subprocess over stdio (or an --ipc socket/pipe).
 //
-// It is a drop-in extraction of the per-service codec that was copy-pasted
-// across ~10 binaries: the wire types (Message, RPCError), the request/
-// notification/response classifiers, and the Codec itself (newline-framed
-// read + concurrency-safe, short-write-safe write). A worker that only sends
-// notifications and responds to requests can adopt this with no behavior
-// change; the bidirectional peer/correlation layer (a child originating
-// id-bearing requests upward) is built separately on top of these types.
+// The codec owns newline framing, protocol-v2 envelopes and a bounded ordered
+// output queue. Notify and responses return acceptance, not guaranteed delivery;
+// Flush/Drain/Close establish delivery barriers. Peer adds request correlation.
+// See README.md for transport ownership, sizing measurements and shutdown rules.
 package jsonrpc
 
 import (
@@ -86,13 +83,7 @@ func (m *Message) IsResponse() bool {
 type Codec struct {
 	scanner *bufio.Scanner
 	writer  io.Writer
-	// A channel semaphore lets canceled requests stop waiting for the write
-	// lock without releasing the lock held by an unfinished frame.
-	wmu chan struct{}
-
-	// A failed write may have emitted a partial frame. Never append another
-	// frame to that stream; only the transport owner can replace it.
-	writeErr error
+	out     orderedWriter
 
 	// seq is this connection's notification counter, starting at 1. The broker
 	// builds one Codec per client connection, so it is already per-connection.
@@ -103,11 +94,9 @@ type Codec struct {
 func NewCodec(rw io.ReadWriter) *Codec {
 	scanner := bufio.NewScanner(rw)
 	scanner.Buffer(make([]byte, 0, 1024*1024), 1024*1024)
-	return &Codec{
-		scanner: scanner,
-		writer:  rw,
-		wmu:     make(chan struct{}, 1),
-	}
+	c := &Codec{scanner: scanner, writer: rw}
+	c.initWriter()
+	return c
 }
 
 // NewCodecMaxFrame is like NewCodec but caps a single inbound frame at
@@ -119,43 +108,9 @@ func NewCodec(rw io.ReadWriter) *Codec {
 func NewCodecMaxFrame(rw io.ReadWriter, maxFrame int) *Codec {
 	scanner := bufio.NewScanner(rw)
 	scanner.Buffer(make([]byte, 0, 64*1024), maxFrame)
-	return &Codec{
-		scanner: scanner,
-		writer:  rw,
-		wmu:     make(chan struct{}, 1),
-	}
-}
-
-// BindWriteContext bounds every write on this connection by ctx cancellation.
-// Install once, before any concurrent codec use. The codec write lock still owns wire
-// ordering and stamps. If Close cannot unblock an OS write, at most that single
-// write is abandoned: subsequent writes fail without spawning another writer.
-// Default codecs retain synchronous write semantics.
-func (c *Codec) BindWriteContext(ctx context.Context) {
-	c.writer = contextWriter{ctx: ctx, writer: c.writer}
-}
-
-type contextWriter struct {
-	ctx    context.Context
-	writer io.Writer
-}
-
-func (w contextWriter) Write(data []byte) (int, error) {
-	if err := w.ctx.Err(); err != nil {
-		return 0, err
-	}
-	type outcome struct {
-		n   int
-		err error
-	}
-	done := make(chan outcome, 1)
-	go func() { n, err := w.writer.Write(data); done <- outcome{n, err} }()
-	select {
-	case result := <-done:
-		return result.n, result.err
-	case <-w.ctx.Done():
-		return 0, w.ctx.Err()
-	}
+	c := &Codec{scanner: scanner, writer: rw}
+	c.initWriter()
+	return c
 }
 
 // Read returns the next frame, or io.EOF when the stream closes. A malformed
@@ -164,7 +119,17 @@ func (w contextWriter) Write(data []byte) (int, error) {
 // which is terminal; a terminal scanner/transport error is returned as a plain
 // error.
 func (c *Codec) Read() (*Message, error) {
+	select {
+	case <-c.Done():
+		return nil, io.EOF
+	default:
+	}
 	if !c.scanner.Scan() {
+		select {
+		case <-c.Done():
+			return nil, io.EOF
+		default:
+		}
 		if err := c.scanner.Err(); err != nil {
 			return nil, fmt.Errorf("read error: %w", err)
 		}
@@ -190,53 +155,9 @@ func (c *Codec) Read() (*Message, error) {
 	return &msg, nil
 }
 
-func (c *Codec) write(msg *Message) error {
-	c.wmu <- struct{}{}
-	defer func() { <-c.wmu }()
-	return c.writeLocked(msg)
-}
-
-// writeNotification stamps a notification frame and writes it as ONE critical
-// section. Stamping before taking wmu would let two concurrent Notify calls
-// invert: A takes seq 1, B takes seq 2, B wins wmu and puts seq 2 on the wire
-// first, violating the sender's monotonic wire guarantee. Receivers accept
-// gaps and regressions without sequence-based recovery.
-func (c *Codec) writeNotification(msg *Message) error {
-	c.wmu <- struct{}{}
-	defer func() { <-c.wmu }()
-	c.stamp(msg)
-	return c.writeLocked(msg)
-}
-
-// writeLocked marshals and writes one frame. The caller must hold wmu.
-func (c *Codec) writeLocked(msg *Message) error {
-	if c.writeErr != nil {
-		return c.writeErr
-	}
-	data, err := json.Marshal(msg)
-	if err != nil {
-		return err
-	}
-	data = append(data, '\n')
-	// Loop on Write so a short-write doesn't truncate the JSON-RPC frame
-	// the peer's bufio.Scanner is about to parse — a half-frame would either
-	// fail to decode or silently merge with the next frame's bytes.
-	for off := 0; off < len(data); {
-		n, werr := c.writer.Write(data[off:])
-		if werr != nil {
-			c.writeErr = werr
-			return werr
-		}
-		if n == 0 {
-			c.writeErr = io.ErrShortWrite
-			return c.writeErr
-		}
-		off += n
-	}
-	return nil
-}
-
-// Notify sends a fire-and-forget notification (method, params, no id).
+// Notify accepts a fire-and-forget notification (method, params, no id).
+// It waits at most WriteAdmissionWait for delivery, then returns acceptance.
+// Use Flush or Close when delivery must precede shutdown.
 func (c *Codec) Notify(method string, params any) error {
 	raw, err := json.Marshal(params)
 	if err != nil {
@@ -258,26 +179,9 @@ func (c *Codec) Request(id *json.RawMessage, method string, params json.RawMessa
 	return c.request(context.Background(), id, method, params)
 }
 
-// request stops waiting for wmu when its caller is canceled. Once a frame
-// starts, keep wmu through the entire write, even if ctx expires: releasing it
-// early would let another writer interleave with a partially written frame.
+// request accepts an unstamped request into the same ordered writer queue.
 func (c *Codec) request(ctx context.Context, id *json.RawMessage, method string, params json.RawMessage) error {
-	select {
-	case <-ctx.Done():
-		return ctx.Err()
-	case c.wmu <- struct{}{}:
-	}
-	defer func() { <-c.wmu }()
-	// Cancellation and lock availability can become ready together.
-	if err := ctx.Err(); err != nil {
-		return err
-	}
-	return c.writeLocked(&Message{
-		JSONRPC: "2.0",
-		ID:      id,
-		Method:  method,
-		Params:  params,
-	})
+	return c.enqueue(ctx, &Message{JSONRPC: "2.0", ID: id, Method: method, Params: params})
 }
 
 // Respond sends a successful response carrying result for the given request id.
diff --git a/services/shared/jsonrpc/jsonrpc_test.go b/services/shared/jsonrpc/jsonrpc_test.go
index f1b59b524..3ceddde4a 100644
--- a/services/shared/jsonrpc/jsonrpc_test.go
+++ b/services/shared/jsonrpc/jsonrpc_test.go
@@ -21,6 +21,13 @@ type rw struct {
 func writeCodec(w io.Writer) *Codec { return NewCodec(rw{Reader: bytes.NewReader(nil), Writer: w}) }
 func readCodec(r io.Reader) *Codec  { return NewCodec(rw{Reader: r, Writer: io.Discard}) }
 
+func mustFlush(t *testing.T, c *Codec) {
+	t.Helper()
+	if err := c.Flush(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+}
+
 func TestClassifiers(t *testing.T) {
 	id := json.RawMessage("1")
 	req := &Message{JSONRPC: "2.0", ID: &id, Method: "m"}
@@ -40,9 +47,11 @@ func TestClassifiers(t *testing.T) {
 
 func TestNotifyRoundTrip(t *testing.T) {
 	var buf bytes.Buffer
-	if err := writeCodec(&buf).Notify("node/discovered", map[string]string{"id": "n1"}); err != nil {
+	sender1 := writeCodec(&buf)
+	if err := sender1.Notify("node/discovered", map[string]string{"id": "n1"}); err != nil {
 		t.Fatalf("Notify: %v", err)
 	}
+	mustFlush(t, sender1)
 	if !bytes.HasSuffix(buf.Bytes(), []byte("\n")) {
 		t.Fatalf("frame not newline-terminated: %q", buf.String())
 	}
@@ -63,18 +72,22 @@ func TestRespondAndError(t *testing.T) {
 	id := json.RawMessage("7")
 
 	var okBuf bytes.Buffer
-	if err := writeCodec(&okBuf).Respond(&id, map[string]int{"level": 1}); err != nil {
+	sender1 := writeCodec(&okBuf)
+	if err := sender1.Respond(&id, map[string]int{"level": 1}); err != nil {
 		t.Fatalf("Respond: %v", err)
 	}
+	mustFlush(t, sender1)
 	msg, err := readCodec(&okBuf).Read()
 	if err != nil || !msg.IsResponse() || msg.Error != nil {
 		t.Fatalf("bad ok response: %+v err=%v", msg, err)
 	}
 
 	var errBuf bytes.Buffer
-	if err := writeCodec(&errBuf).RespondError(&id, -32601, "method not found"); err != nil {
+	sender2 := writeCodec(&errBuf)
+	if err := sender2.RespondError(&id, -32601, "method not found"); err != nil {
 		t.Fatalf("RespondError: %v", err)
 	}
+	mustFlush(t, sender2)
 	emsg, err := readCodec(&errBuf).Read()
 	if err != nil {
 		t.Fatalf("Read err response: %v", err)
@@ -98,9 +111,11 @@ func (w *oneByteWriter) Write(p []byte) (int, error) {
 
 func TestShortWriteWritesFullFrame(t *testing.T) {
 	var w oneByteWriter
-	if err := writeCodec(&w).Notify("m", map[string]string{"k": "value-that-spans-many-writes"}); err != nil {
+	sender1 := writeCodec(&w)
+	if err := sender1.Notify("m", map[string]string{"k": "value-that-spans-many-writes"}); err != nil {
 		t.Fatalf("Notify: %v", err)
 	}
+	mustFlush(t, sender1)
 	msg, err := readCodec(bytes.NewReader(w.buf.Bytes())).Read()
 	if err != nil {
 		t.Fatalf("Read after short writes: %v", err)
@@ -135,9 +150,11 @@ func TestReadMalformedFrameIsRecoverableDecodeError(t *testing.T) {
 func TestRespondErrorDataRoundTrip(t *testing.T) {
 	id := json.RawMessage("3")
 	var buf bytes.Buffer
-	if err := writeCodec(&buf).RespondErrorData(&id, -32602, "bad", map[string]string{"field": "port"}); err != nil {
+	sender1 := writeCodec(&buf)
+	if err := sender1.RespondErrorData(&id, -32602, "bad", map[string]string{"field": "port"}); err != nil {
 		t.Fatalf("RespondErrorData: %v", err)
 	}
+	mustFlush(t, sender1)
 	msg, err := readCodec(&buf).Read()
 	if err != nil {
 		t.Fatalf("Read: %v", err)
@@ -151,7 +168,9 @@ func TestRespondErrorDataRoundTrip(t *testing.T) {
 	}
 	// nil data omits the field entirely.
 	var buf2 bytes.Buffer
-	_ = writeCodec(&buf2).RespondErrorData(&id, -32603, "boom", nil)
+	sender2 := writeCodec(&buf2)
+	_ = sender2.RespondErrorData(&id, -32603, "boom", nil)
+	mustFlush(t, sender2)
 	m2, _ := readCodec(&buf2).Read()
 	if len(m2.Error.Data) != 0 {
 		t.Fatalf("expected empty data for nil, got %q", string(m2.Error.Data))
@@ -161,9 +180,11 @@ func TestRespondErrorDataRoundTrip(t *testing.T) {
 func TestNewCodecMaxFrameAcceptsLargeFrame(t *testing.T) {
 	big := strings.Repeat("x", 2<<20) // 2 MiB, over the 1 MiB default
 	var buf bytes.Buffer
-	if err := writeCodec(&buf).Notify("m", map[string]string{"blob": big}); err != nil {
+	sender1 := writeCodec(&buf)
+	if err := sender1.Notify("m", map[string]string{"blob": big}); err != nil {
 		t.Fatalf("Notify: %v", err)
 	}
+	mustFlush(t, sender1)
 	c := NewCodecMaxFrame(rw{Reader: bytes.NewReader(buf.Bytes()), Writer: io.Discard}, 8<<20)
 	msg, err := c.Read()
 	if err != nil {
diff --git a/services/shared/jsonrpc/peer.go b/services/shared/jsonrpc/peer.go
index 6edb582e7..50c616d13 100644
--- a/services/shared/jsonrpc/peer.go
+++ b/services/shared/jsonrpc/peer.go
@@ -28,21 +28,18 @@ var ErrPeerClosed = errors.New("jsonrpc peer closed")
 // lets a child originate an upward Call without deadlocking its own read loop
 // (the read pump lives in Serve, separate from the Call goroutine).
 //
-// Peer does NOT own the process/transport lifecycle: the caller builds the
-// Codec (e.g. over a child's stdout/stdin), runs Serve in a goroutine, and
-// calls Close when the transport goes away. Serve also closes the peer when
-// Read returns a terminal error, so in-flight and future Calls fail fast.
+// The caller owns the process lifecycle and supplies an interruptible transport.
+// Close and terminal read/write errors abort the codec and fail pending calls.
+// For a clean control-pipe EOF, Drain the codec before closing that pipe and
+// keep the child stdout drained through the process owner's bounded join.
 type Peer struct {
 	codec *Codec
 
 	idCounter atomic.Int64
 
-	mu      sync.Mutex
-	pending map[int64]chan *Message
-	closed  bool
-	// Tail of the relay write queue; responses never hold up later writes.
-	relayWriteDone <-chan struct{}
-
+	mu           sync.Mutex
+	pending      map[int64]chan *Message
+	closed       bool
 	closeOnce    sync.Once
 	closedCh     chan struct{}
 	relayCtx     context.Context
@@ -90,8 +87,8 @@ func (p *Peer) RespondErrorData(id *json.RawMessage, code int, message string, d
 // the peer reported separately; a non-nil error means the call itself couldn't
 // complete (transport gone, ctx cancelled/deadline). Callers wanting a fixed
 // timeout should pass a ctx with a deadline. Cancellation does not interrupt an
-// in-flight frame: its write keeps the codec lock until it finishes or the
-// transport owner closes the writer. Queued writes stop waiting on cancellation.
+// in-flight frame: the single ordered writer finishes it or kills the stream
+// at its write deadline. Canceled requests not yet started are skipped.
 func (p *Peer) Call(ctx context.Context, method string, params json.RawMessage) (json.RawMessage, *RPCError, error) {
 	if err := ctx.Err(); err != nil {
 		return nil, nil, err
@@ -113,6 +110,8 @@ func (p *Peer) Call(ctx context.Context, method string, params json.RawMessage)
 		return nil, nil, ctx.Err()
 	case <-p.closedCh:
 		return nil, nil, ErrPeerClosed
+	case <-p.codec.Done():
+		return nil, nil, p.codec.Err()
 	case err := <-written:
 		if ctx.Err() != nil {
 			return nil, nil, ctx.Err()
@@ -127,6 +126,8 @@ func (p *Peer) Call(ctx context.Context, method string, params json.RawMessage)
 		return nil, nil, ctx.Err()
 	case <-p.closedCh:
 		return nil, nil, ErrPeerClosed
+	case <-p.codec.Done():
+		return nil, nil, p.codec.Err()
 	case resp := <-ch:
 		return resp.Result, resp.Error, nil
 	}
@@ -138,55 +139,38 @@ func (p *Peer) Call(ctx context.Context, method string, params json.RawMessage)
 // via push events). Both writing and responding run off the caller's loop.
 // Relay writes retain submission order. Returns an error only if the peer is
 // already closed; accepted requests report write failures through respond.
-// Close cancels queued writes and fails pending relays; the transport owner
-// must close the writer to release any in-flight frame (as with Call).
+// Close cancels queued writes, fails pending relays and aborts the transport.
 func (p *Peer) RelayRequest(method string, params json.RawMessage, respond func(result json.RawMessage, rpcErr *RPCError, err error)) error {
 	ch, id, err := p.register()
 	if err != nil {
 		return err
 	}
 
-	// Reserve queue position synchronously: simply launching write goroutines
-	// would let later engine commands overtake earlier ones at the codec lock.
-	p.mu.Lock()
-	previous := p.relayWriteDone
-	done := make(chan struct{})
-	p.relayWriteDone = done
-	p.mu.Unlock()
-	params = bytes.Clone(params)
-	ctx := p.relayCtx
-	written := make(chan error, 1)
-	go func() {
-		defer close(done)
-		if previous != nil {
-			select {
-			case <-previous:
-			case <-ctx.Done():
-				return
-			}
-		}
-		written <- p.writeRequest(ctx, id, method, params)
-	}()
+	// Submit synchronously into the codec FIFO, not a second queue of blocked
+	// goroutines: otherwise a later Notify/Respond could overtake this request.
+	idRaw := json.RawMessage(strconv.FormatInt(id, 10))
+	written, writeErr := p.codec.submit(p.relayCtx, &Message{JSONRPC: "2.0", ID: &idRaw, Method: method, Params: params})
 
 	go func() {
 		defer p.clearPending(id)
+		if writeErr != nil {
+			respond(nil, nil, writeErr)
+			return
+		}
 		select {
 		case <-p.closedCh:
 			respond(nil, nil, ErrPeerClosed)
 			return
-		case err := <-written:
-			if ctx.Err() != nil {
-				respond(nil, nil, ErrPeerClosed)
-				return
-			}
-			if err != nil {
-				respond(nil, nil, err)
-				return
-			}
+		case <-p.codec.Done():
+			respond(nil, nil, p.codec.Err())
+			return
+		case <-written:
 		}
 		select {
 		case <-p.closedCh:
 			respond(nil, nil, ErrPeerClosed)
+		case <-p.codec.Done():
+			respond(nil, nil, p.codec.Err())
 		case resp := <-ch:
 			respond(resp.Result, resp.Error, nil)
 		}
@@ -233,7 +217,8 @@ func (p *Peer) deliverResponse(msg *Message) {
 	}
 }
 
-// Close marks the peer closed and wakes all pending Calls with ErrPeerClosed.
+// Close aborts the peer and wakes all pending Calls with ErrPeerClosed.
+// Drain the codec first when a clean shutdown must deliver accepted frames.
 // Idempotent; safe to call from Serve and from a process-exit watcher.
 func (p *Peer) Close() {
 	p.closeOnce.Do(func() {
@@ -242,6 +227,7 @@ func (p *Peer) Close() {
 		p.mu.Unlock()
 		close(p.closedCh)
 		p.cancelRelays()
+		p.codec.terminate(ErrPeerClosed)
 	})
 }
 
diff --git a/services/shared/jsonrpc/peer_test.go b/services/shared/jsonrpc/peer_test.go
index 71c9ce317..89cc0f6fa 100644
--- a/services/shared/jsonrpc/peer_test.go
+++ b/services/shared/jsonrpc/peer_test.go
@@ -263,11 +263,8 @@ func TestCanceledQueuedRequestWritersExit(t *testing.T) {
 			t.Fatalf("only %d/%d canceled queued writers exited while the first frame was blocked", i, queued)
 		}
 	}
-	select {
-	case err := <-first:
-		t.Fatalf("first writer finished before its frame was drained: %v", err)
-	default:
-	}
+	// Notify now returns on admission; the first frame is still owned by
+	// the single writer even after its producer returns.
 
 	// Cancellation must not release the active writer's lock or damage its
 	// frame. Let it finish only after every canceled queued writer has exited.
@@ -338,8 +335,11 @@ func TestCallBlockedWriteFinishesWhenWriterCloses(t *testing.T) {
 		t.Fatal("Close did not release Call with a blocked write")
 	}
 	writer.Close()
-	// Acquiring the codec lock proves the abandoned write has finished.
-	go func() { peer.codec.wmu <- struct{}{}; <-peer.codec.wmu; done <- nil }()
+	// Joining the writer proves the interrupted write has finished.
+	peer.codec.out.mu.Lock()
+	idle := peer.codec.out.idle
+	peer.codec.out.mu.Unlock()
+	go func() { <-idle; done <- nil }()
 	select {
 	case <-done:
 	case <-time.After(time.Second):
diff --git a/services/shared/jsonrpc/relay_order_test.go b/services/shared/jsonrpc/relay_order_test.go
new file mode 100644
index 000000000..c8a183875
--- /dev/null
+++ b/services/shared/jsonrpc/relay_order_test.go
@@ -0,0 +1,57 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package jsonrpc
+
+import (
+	"bytes"
+	"encoding/json"
+	"io"
+	"testing"
+	"time"
+)
+
+func TestRelayInvalidParamsReportsSubmissionError(t *testing.T) {
+	c, _ := leakCheckedPipe(t)
+	peer := NewPeer(c)
+	defer peer.Close()
+	done := make(chan error, 1)
+	if err := peer.RelayRequest("invalid", json.RawMessage(`{`), func(_ json.RawMessage, _ *RPCError, err error) { done <- err }); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case err := <-done:
+		if err == nil {
+			t.Fatal("invalid params succeeded")
+		}
+	case <-time.After(100 * time.Millisecond):
+		t.Fatal("submission error stranded relay callback")
+	}
+}
+
+func TestRelayAdmissionOrdersWithNotifications(t *testing.T) {
+	c, remote := leakCheckedPipe(t)
+	peer := NewPeer(c)
+	defer peer.Close()
+	ignore := func(json.RawMessage, *RPCError, error) {}
+	if err := peer.RelayRequest("first", nil, ignore); err != nil {
+		t.Fatal(err)
+	}
+	prefix := make([]byte, 1)
+	if _, err := io.ReadFull(remote, prefix); err != nil {
+		t.Fatal(err)
+	}
+	if err := peer.RelayRequest("second", nil, ignore); err != nil {
+		t.Fatal(err)
+	}
+	if err := peer.Notify("third", nil); err != nil {
+		t.Fatal(err)
+	}
+	reader := readCodec(io.MultiReader(bytes.NewReader(prefix), remote))
+	for _, want := range []string{"first", "second", "third"} {
+		msg, err := reader.Read()
+		if err != nil || msg.Method != want {
+			t.Fatalf("want %s, got %+v, %v", want, msg, err)
+		}
+	}
+}
diff --git a/services/shared/jsonrpc/stdio.go b/services/shared/jsonrpc/stdio.go
new file mode 100644
index 000000000..269358731
--- /dev/null
+++ b/services/shared/jsonrpc/stdio.go
@@ -0,0 +1,32 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package jsonrpc
+
+import (
+	"errors"
+	"io"
+	"os"
+)
+
+type stdioStream struct{ reader, writer *os.File }
+
+func (s *stdioStream) Read(p []byte) (int, error)  { return s.reader.Read(p) }
+func (s *stdioStream) Write(p []byte) (int, error) { return s.writer.Write(p) }
+func (s *stdioStream) Close() error                { return errors.Join(s.writer.Close(), s.reader.Close()) }
+
+// NewStdioTransport takes ownership of stdin/stdout. On Unix inherited pipes
+// must be registered with the Go poller so Close can interrupt a full-pipe write.
+// Do not use os.Stdin/os.Stdout after this call (stderr is unaffected).
+func NewStdioTransport() (io.ReadWriteCloser, error) {
+	reader, err := interruptibleStdioFile(os.Stdin)
+	if err != nil {
+		return nil, err
+	}
+	writer, err := interruptibleStdioFile(os.Stdout)
+	if err != nil {
+		_ = reader.Close()
+		return nil, err
+	}
+	return &stdioStream{reader: reader, writer: writer}, nil
+}
diff --git a/services/shared/jsonrpc/stdio_test.go b/services/shared/jsonrpc/stdio_test.go
new file mode 100644
index 000000000..d20fd09bc
--- /dev/null
+++ b/services/shared/jsonrpc/stdio_test.go
@@ -0,0 +1,47 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package jsonrpc
+
+import (
+	"errors"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"go.uber.org/goleak"
+)
+
+func TestInheritedStdioPipeDeadlineJoinsWriter(t *testing.T) {
+	baseline := goleak.IgnoreCurrent()
+	t.Cleanup(func() { goleak.VerifyNone(t, baseline) })
+	input, inputWriter, err := os.Pipe()
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = input.Close(); _ = inputWriter.Close() })
+	outputReader, output, err := os.Pipe()
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = outputReader.Close(); _ = output.Close() })
+	_ = output.Fd() // reproduce an inherited, blocking Unix stdout descriptor
+	pollable, err := interruptibleStdioFile(output)
+	if err != nil {
+		t.Fatal(err)
+	}
+	c := NewCodec(&stdioStream{reader: input, writer: pollable})
+	t.Cleanup(func() { _ = c.Close() })
+	if err := c.Notify("large-snapshot", strings.Repeat("x", 1<<20)); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-c.Done():
+	case <-time.After(WriteTimeout + time.Second):
+		t.Fatal("inherited full pipe defeated deadline")
+	}
+	if !errors.Is(c.Err(), ErrWriteTimeout) {
+		t.Fatalf("terminal reason: %v", c.Err())
+	}
+}
diff --git a/services/shared/jsonrpc/stdio_unix.go b/services/shared/jsonrpc/stdio_unix.go
new file mode 100644
index 000000000..80c608dae
--- /dev/null
+++ b/services/shared/jsonrpc/stdio_unix.go
@@ -0,0 +1,26 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+//go:build !windows
+
+package jsonrpc
+
+import (
+	"golang.org/x/sys/unix"
+	"os"
+)
+
+func interruptibleStdioFile(f *os.File) (*os.File, error) {
+	fd, err := unix.FcntlInt(f.Fd(), unix.F_DUPFD_CLOEXEC, 0)
+	if err != nil {
+		return nil, err
+	}
+	if err := unix.SetNonblock(fd, true); err != nil {
+		_ = unix.Close(fd)
+		return nil, err
+	}
+	// NewFile recognizes O_NONBLOCK and registers pipes with the runtime poller.
+	replacement := os.NewFile(uintptr(fd), f.Name())
+	_ = f.Close() // no original descriptor may keep the dead stream alive
+	return replacement, nil
+}
diff --git a/services/shared/jsonrpc/stdio_windows.go b/services/shared/jsonrpc/stdio_windows.go
new file mode 100644
index 000000000..592da6c27
--- /dev/null
+++ b/services/shared/jsonrpc/stdio_windows.go
@@ -0,0 +1,9 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package jsonrpc
+
+import "os"
+
+// os.File.Close cancels pending pipe I/O on Windows.
+func interruptibleStdioFile(f *os.File) (*os.File, error) { return f, nil }
diff --git a/services/shared/jsonrpc/writer.go b/services/shared/jsonrpc/writer.go
new file mode 100644
index 000000000..e1794f837
--- /dev/null
+++ b/services/shared/jsonrpc/writer.go
@@ -0,0 +1,330 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package jsonrpc
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"log/slog"
+	"sync"
+	"time"
+)
+
+// Limits include the in-flight frame. See README.md for burst measurements.
+const (
+	WriteQueueFrames = 256
+	WriteQueueBytes  = 16 << 20
+	WriteTimeout     = 5 * time.Second
+	// Wait briefly for the common immediately-written case, never for a pipe.
+	WriteAdmissionWait = 10 * time.Millisecond
+)
+
+var (
+	ErrWriteQueueFull = errors.New("jsonrpc write queue full")
+	ErrWriteTimeout   = errors.New("jsonrpc write deadline exceeded")
+	ErrCodecClosed    = errors.New("jsonrpc codec closed")
+)
+
+type queuedFrame struct {
+	data     []byte
+	deadline time.Time
+	done     chan struct{}
+	ctx      context.Context
+}
+
+type orderedWriter struct {
+	mu          sync.Mutex
+	queue       []queuedFrame
+	bytes       int
+	frames      int
+	running     bool
+	closing     bool
+	err         error
+	done        chan struct{}
+	idle        chan struct{}
+	tail        <-chan struct{}
+	cleanupDone chan struct{}
+	closeOnce   sync.Once
+	stopContext func() bool
+}
+
+func (c *Codec) initWriter() {
+	c.out.done = make(chan struct{})
+	c.out.idle = make(chan struct{})
+	close(c.out.idle)
+	c.out.tail = c.out.idle
+	c.out.cleanupDone = make(chan struct{})
+}
+
+// BindWriteContext terminates the stream on cancellation. Install before use.
+// Cancellation is abortive; for a clean shutdown call Close before canceling.
+func (c *Codec) BindWriteContext(ctx context.Context) {
+	c.out.stopContext = context.AfterFunc(ctx, func() { c.terminate(ctx.Err()) })
+}
+
+// Done closes on terminal write failure or Close, including queue overflow.
+func (c *Codec) Done() <-chan struct{} { return c.out.done }
+
+// Err reports the terminal stream reason, or nil while it is live.
+func (c *Codec) Err() error { c.out.mu.Lock(); defer c.out.mu.Unlock(); return c.out.err }
+
+func (c *Codec) write(msg *Message) error             { return c.enqueue(context.Background(), msg) }
+func (c *Codec) writeNotification(msg *Message) error { return c.write(msg) }
+
+func (c *Codec) submit(ctx context.Context, msg *Message) (<-chan struct{}, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	c.out.mu.Lock()
+	if c.out.err != nil || c.out.closing {
+		err := c.out.err
+		if err == nil {
+			err = ErrCodecClosed
+		}
+		c.out.mu.Unlock()
+		return nil, err
+	}
+	if err := ctx.Err(); err != nil {
+		c.out.mu.Unlock()
+		return nil, err
+	}
+	if msg.IsNotification() {
+		c.stamp(msg)
+	}
+	data, err := json.Marshal(msg)
+	if err != nil {
+		c.out.mu.Unlock()
+		return nil, err
+	}
+	data = append(data, '\n')
+	if c.out.frames >= WriteQueueFrames || len(data) > WriteQueueBytes-c.out.bytes {
+		// Set terminal state before releasing the enqueue lock: no later frame
+		// may be accepted after the rejected notification consumed a sequence.
+		c.failLocked(ErrWriteQueueFull)
+		c.out.mu.Unlock()
+		go c.finishTermination(ErrWriteQueueFull)
+		return nil, ErrWriteQueueFull
+	}
+	if !c.out.running {
+		c.out.running = true
+		c.out.idle = make(chan struct{})
+	}
+	start := c.out.frames == 0
+	frame := queuedFrame{data: data, deadline: time.Now().Add(WriteTimeout), done: make(chan struct{}), ctx: ctx}
+	c.out.queue = append(c.out.queue, frame)
+	c.out.tail = frame.done
+	c.out.frames++
+	c.out.bytes += len(data)
+	c.out.mu.Unlock()
+	if start {
+		go c.drain()
+	}
+	return frame.done, nil
+}
+
+func (c *Codec) enqueue(ctx context.Context, msg *Message) error {
+	done, err := c.submit(ctx, msg)
+	if err != nil {
+		return err
+	}
+	// Acceptance is not delivery. The brief completion fast path avoids a
+	// scheduler round trip for shutdown acknowledgements, but a stalled reader
+	// never keeps a producer (including a lock holder) here beyond this wait.
+	wait := WriteAdmissionWait
+	if msg.IsRequest() {
+		wait = WriteTimeout
+	}
+	timer := time.NewTimer(wait)
+	defer timer.Stop()
+	select {
+	case <-done:
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		c.out.mu.Lock()
+		err := c.out.err
+		c.out.mu.Unlock()
+		return err
+	case <-c.out.done:
+		c.out.mu.Lock()
+		err := c.out.err
+		c.out.mu.Unlock()
+		return err
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-timer.C:
+		if msg.IsRequest() {
+			c.terminate(ErrWriteTimeout)
+			return ErrWriteTimeout
+		}
+		return nil
+	}
+}
+
+// failLocked never logs an underlying error: a custom transport's error can
+// contain message contents. Only these fixed transport categories reach logs.
+func (c *Codec) failLocked(err error) {
+	c.out.err = err
+	c.out.queue = nil
+	close(c.out.done)
+}
+
+func (c *Codec) terminate(err error) {
+	c.out.mu.Lock()
+	first := c.out.err == nil
+	if first {
+		c.failLocked(err)
+	}
+	c.out.mu.Unlock()
+	if first {
+		// Neither transport Close nor a stderr logger may run on a producer's
+		// lock-holding path. Only one terminal cleanup goroutine is ever started.
+		go c.finishTermination(err)
+	}
+}
+
+func (c *Codec) finishTermination(err error) {
+	defer close(c.out.cleanupDone)
+	c.closeStream()
+	reason := "write failed"
+	if errors.Is(err, ErrWriteQueueFull) {
+		reason = "queue full"
+	}
+	if errors.Is(err, ErrWriteTimeout) {
+		reason = "write deadline"
+	}
+	if !errors.Is(err, ErrCodecClosed) && !errors.Is(err, ErrPeerClosed) && !errors.Is(err, context.Canceled) {
+		slog.Warn("JSON-RPC stream terminated", "reason", reason)
+	}
+}
+
+func (c *Codec) closeStream() {
+	c.out.closeOnce.Do(func() {
+		if closer, ok := c.writer.(io.Closer); ok {
+			_ = closer.Close()
+		}
+	})
+}
+
+func (c *Codec) drain() {
+	for {
+		c.out.mu.Lock()
+		if c.out.err != nil || len(c.out.queue) == 0 {
+			c.out.running = false
+			close(c.out.idle)
+			c.out.mu.Unlock()
+			return
+		}
+		frame := c.out.queue[0]
+		c.out.queue[0] = queuedFrame{}
+		c.out.queue = c.out.queue[1:]
+		c.out.mu.Unlock()
+		timer := time.AfterFunc(time.Until(frame.deadline), func() { c.terminate(ErrWriteTimeout) })
+		var err error
+		if frame.ctx.Err() == nil {
+			err = writeFrame(c.writer, frame.data)
+		}
+		timer.Stop()
+		if err != nil {
+			c.terminate(err)
+			c.out.mu.Lock()
+			c.out.running = false
+			close(c.out.idle)
+			c.out.mu.Unlock()
+			return
+		}
+		c.out.mu.Lock()
+		c.out.bytes -= len(frame.data)
+		c.out.frames--
+		close(frame.done)
+		// Transfer idle ownership under the enqueue lock, so a producer cannot
+		// append to an empty queue while this goroutine is exiting.
+		if c.out.frames == 0 {
+			c.out.running = false
+			close(c.out.idle)
+			c.out.mu.Unlock()
+			return
+		}
+		c.out.mu.Unlock()
+	}
+}
+
+func writeFrame(w io.Writer, data []byte) error {
+	for len(data) > 0 {
+		n, err := w.Write(data)
+		if err != nil {
+			return err
+		}
+		if n <= 0 || n > len(data) {
+			return io.ErrShortWrite
+		}
+		data = data[n:]
+	}
+	return nil
+}
+
+// Flush waits for frames accepted before this call, never longer than
+// WriteTimeout (or ctx). A timeout kills the stream, not just the waiter.
+func (c *Codec) Flush(ctx context.Context) error {
+	c.out.mu.Lock()
+	if c.out.err != nil {
+		err := c.out.err
+		c.out.mu.Unlock()
+		return err
+	}
+	done := c.out.tail
+	c.out.mu.Unlock()
+	timer := time.NewTimer(WriteTimeout)
+	defer timer.Stop()
+	select {
+	case <-c.out.done:
+	case <-done:
+		c.out.mu.Lock()
+		err := c.out.err
+		c.out.mu.Unlock()
+		return err
+	case <-ctx.Done():
+		c.terminate(ctx.Err())
+	case <-timer.C:
+		c.terminate(ErrWriteTimeout)
+	}
+	c.out.mu.Lock()
+	err := c.out.err
+	c.out.mu.Unlock()
+	return err
+}
+
+// Drain stops admission and flushes accepted frames, leaving the read half
+// open for a process owner's cleanup join. The owner must then close control.
+func (c *Codec) Drain(ctx context.Context) error {
+	c.out.mu.Lock()
+	c.out.closing = true
+	c.out.mu.Unlock()
+	return c.Flush(ctx)
+}
+
+// Close stops admission, flushes accepted frames within one total WriteTimeout,
+// then closes the transport. Transports must implement Close that interrupts
+// Read/Write. Non-closeable in-memory codecs are useful only for finite writes.
+func (c *Codec) Close() error {
+	deadline := time.Now().Add(WriteTimeout)
+	err := c.Drain(context.Background())
+	c.terminate(ErrCodecClosed)
+	c.closeStream()
+	// Join terminal logging too, without adding a second shutdown allowance.
+	if remaining := time.Until(deadline); remaining > 0 {
+		timer := time.NewTimer(remaining)
+		select {
+		case <-c.out.cleanupDone:
+		case <-timer.C:
+		}
+		timer.Stop()
+	}
+	if c.out.stopContext != nil {
+		c.out.stopContext()
+	}
+	return err
+}
diff --git a/services/shared/jsonrpc/writer_test.go b/services/shared/jsonrpc/writer_test.go
new file mode 100644
index 000000000..2c35efeb5
--- /dev/null
+++ b/services/shared/jsonrpc/writer_test.go
@@ -0,0 +1,329 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package jsonrpc
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"log/slog"
+	"net"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"go.uber.org/goleak"
+)
+
+func leakCheckedPipe(t *testing.T) (*Codec, net.Conn) {
+	t.Helper()
+	baseline := goleak.IgnoreCurrent()
+	t.Cleanup(func() { goleak.VerifyNone(t, baseline) })
+	local, remote := net.Pipe()
+	c := NewCodec(local)
+	t.Cleanup(func() { _ = remote.Close(); _ = c.Close() })
+	return c, remote
+}
+
+func TestNotifyStalledReaderReturnsPromptly(t *testing.T) {
+	c, _ := leakCheckedPipe(t)
+	done := make(chan error, 1)
+	go func() { done <- c.Notify("workers:update", map[string]int{"workers": 12}) }()
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatal(err)
+		}
+	case <-time.After(100 * time.Millisecond):
+		t.Fatal("Notify blocked on a stalled reader")
+	}
+}
+
+func TestResponsesStalledReaderReturnPromptly(t *testing.T) {
+	c, _ := leakCheckedPipe(t)
+	id := json.RawMessage("1")
+	done := make(chan error, 1)
+	go func() { done <- c.Respond(&id, true) }()
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatal(err)
+		}
+	case <-time.After(100 * time.Millisecond):
+		t.Fatal("response blocked request dispatch on a stalled reader")
+	}
+}
+
+func TestWriterConcurrentResponseNotificationOrder(t *testing.T) {
+	c, remote := leakCheckedPipe(t)
+	const producers = 64
+	var wg sync.WaitGroup
+	for i := range producers {
+		wg.Go(func() {
+			id := json.RawMessage(strconv.Itoa(i))
+			if err := c.Notify("before", i); err != nil {
+				t.Error(err)
+				return
+			}
+			if err := c.Respond(&id, i); err != nil {
+				t.Error(err)
+			}
+		})
+	}
+	wg.Wait() // all accepted while the remote deliberately reads nothing
+	reader := NewCodec(remote)
+	seen := make(map[int]bool)
+	seq := uint64(0)
+	for range producers * 2 {
+		msg, err := reader.Read()
+		if err != nil {
+			t.Fatal(err)
+		}
+		if msg.IsNotification() {
+			seq++
+			if msg.Seq == nil || *msg.Seq != seq {
+				t.Fatalf("sequence at %d: %v", seq, msg.Seq)
+			}
+			var i int
+			if err := json.Unmarshal(msg.Params, &i); err != nil {
+				t.Fatal(err)
+			}
+			seen[i] = true
+		} else {
+			var i int
+			if err := json.Unmarshal(msg.Result, &i); err != nil {
+				t.Fatal(err)
+			}
+			if !seen[i] || msg.Epoch != nil || msg.Seq != nil {
+				t.Fatalf("response overtook notification or acquired a stamp: %+v", msg)
+			}
+		}
+	}
+	if err := c.Flush(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestWriterQueueOverflowTerminatesStream(t *testing.T) {
+	c, remote := leakCheckedPipe(t)
+	// The active write and queued frames together consume the admission bound.
+	for range WriteQueueFrames {
+		if err := c.Notify("queued", nil); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := c.Notify("overflow", nil); !errors.Is(err, ErrWriteQueueFull) {
+		t.Fatalf("overflow: %v", err)
+	}
+	select {
+	case <-c.Done():
+	default:
+		t.Fatal("overflow left stream alive")
+	}
+	if err := c.Notify("after-overflow", nil); !errors.Is(err, ErrWriteQueueFull) {
+		t.Fatalf("reused dead stream: %v", err)
+	}
+	_ = remote.SetReadDeadline(time.Now().Add(time.Second))
+	if _, err := io.ReadAll(remote); err != nil {
+		t.Fatalf("overflow did not close stream: %v", err)
+	}
+}
+
+func TestWriterByteOverflowTerminatesStream(t *testing.T) {
+	c, _ := leakCheckedPipe(t)
+	if err := c.Notify("too-large", strings.Repeat("x", WriteQueueBytes)); !errors.Is(err, ErrWriteQueueFull) {
+		t.Fatalf("byte bound: %v", err)
+	}
+	select {
+	case <-c.Done():
+	default:
+		t.Fatal("byte overflow left stream alive")
+	}
+}
+
+func TestWriterDeadlineTerminatesStream(t *testing.T) {
+	c, remote := leakCheckedPipe(t)
+	start := time.Now()
+	if err := c.Notify("stalled", nil); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-c.Done():
+	case <-time.After(WriteTimeout + time.Second):
+		t.Fatal("stalled write outlived deadline")
+	}
+	if time.Since(start) < WriteTimeout {
+		t.Fatal("deadline fired too soon")
+	}
+	if err := c.Notify("later", nil); !errors.Is(err, ErrWriteTimeout) {
+		t.Fatalf("deadline error: %v", err)
+	}
+	_ = remote.SetReadDeadline(time.Now().Add(time.Second))
+	if _, err := io.ReadAll(remote); err != nil {
+		t.Fatalf("deadline did not close stream: %v", err)
+	}
+}
+
+func TestWriterCleanCloseFlushesQueuedFrames(t *testing.T) {
+	c, remote := leakCheckedPipe(t)
+	const frames = 16
+	for range frames {
+		if err := c.Notify("queued", nil); err != nil {
+			t.Fatal(err)
+		}
+	}
+	done := make(chan error, 1)
+	go func() { done <- c.Close() }()
+	reader := NewCodec(remote)
+	for i := range frames {
+		msg, err := reader.Read()
+		if err != nil || msg.Seq == nil || *msg.Seq != uint64(i+1) {
+			t.Fatalf("flush frame %d: %+v, %v", i, msg, err)
+		}
+	}
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatal(err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("clean Close did not finish after flushing")
+	}
+	if _, err := reader.Read(); err != io.EOF {
+		t.Fatalf("Close did not close transport: %v", err)
+	}
+}
+
+func TestWriterCleanCloseIsBoundedWhenReaderStalls(t *testing.T) {
+	c, _ := leakCheckedPipe(t)
+	if err := c.Notify("queued", nil); err != nil {
+		t.Fatal(err)
+	}
+	done := make(chan error, 1)
+	go func() { done <- c.Close() }()
+	select {
+	case err := <-done:
+		if !errors.Is(err, ErrWriteTimeout) {
+			t.Fatalf("Close: %v", err)
+		}
+	case <-time.After(WriteTimeout + time.Second):
+		t.Fatal("Close waited forever to flush")
+	}
+}
+
+type observedFlushContext struct {
+	context.Context
+	once    sync.Once
+	entered chan struct{}
+}
+
+func (c *observedFlushContext) Done() <-chan struct{} {
+	c.once.Do(func() { close(c.entered) })
+	return c.Context.Done()
+}
+
+func TestFlushDoesNotWaitForLaterAdmissions(t *testing.T) {
+	c, remote := leakCheckedPipe(t)
+	if err := c.Notify("before-barrier", nil); err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithTimeout(t.Context(), 500*time.Millisecond)
+	defer cancel()
+	observed := &observedFlushContext{Context: ctx, entered: make(chan struct{})}
+	done := make(chan error, 1)
+	go func() { done <- c.Flush(observed) }()
+	<-observed.entered // Flush has captured its barrier before this admission.
+	if err := c.Notify("after-barrier", nil); err != nil {
+		t.Fatal(err)
+	}
+	msg, err := NewCodec(remote).Read()
+	if err != nil || msg.Method != "before-barrier" {
+		t.Fatalf("first frame: %+v, %v", msg, err)
+	}
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("Flush waited for a later frame: %v", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Flush did not finish at its submission barrier")
+	}
+}
+
+type failingDiagnosticStream struct{}
+
+func (failingDiagnosticStream) Read([]byte) (int, error) { return 0, io.EOF }
+func (failingDiagnosticStream) Write([]byte) (int, error) {
+	return 0, errors.New("private-transport-error")
+}
+func (failingDiagnosticStream) Close() error { return nil }
+
+func TestWriterTerminalDiagnosticOnceWithoutPayloads(t *testing.T) {
+	var logs syncBuffer
+	previous := slog.Default()
+	slog.SetDefault(slog.New(slog.NewTextHandler(&logs, nil)))
+	defer slog.SetDefault(previous)
+	c := NewCodec(failingDiagnosticStream{})
+	_ = c.Notify("private-method", map[string]string{"private-field": "private-value"})
+	_ = c.Notify("later", nil)
+	_ = c.Close() // join the single asynchronous diagnostic
+	output := string(logs.Bytes())
+	if strings.Count(output, "JSON-RPC stream terminated") != 1 {
+		t.Fatalf("terminal log count: %q", output)
+	}
+	for _, private := range []string{"private-transport-error", "private-method", "private-field", "private-value"} {
+		if strings.Contains(output, private) {
+			t.Fatalf("terminal log exposed %s", private)
+		}
+	}
+}
+
+func TestDrainStopsAdmissionAndKeepsReadHalfOpen(t *testing.T) {
+	c, remote := leakCheckedPipe(t)
+	if err := c.Notify("last", nil); err != nil {
+		t.Fatal(err)
+	}
+	observed := &observedFlushContext{Context: t.Context(), entered: make(chan struct{})}
+	done := make(chan error, 1)
+	go func() { done <- c.Drain(observed) }()
+	<-observed.entered
+	if err := c.Notify("too-late", nil); !errors.Is(err, ErrCodecClosed) {
+		t.Fatalf("Drain allowed new admission: %v", err)
+	}
+	if _, err := NewCodec(remote).Read(); err != nil {
+		t.Fatal(err)
+	}
+	if err := <-done; err != nil {
+		t.Fatal(err)
+	}
+	go func() {
+		_, err := io.WriteString(remote, `{"jsonrpc":"2.0","id":1,"method":"still-readable"}`+"\n")
+		done <- err
+	}()
+	msg, err := c.Read()
+	if err != nil || msg.Method != "still-readable" {
+		t.Fatalf("Drain closed read half: %+v, %v", msg, err)
+	}
+	if err := <-done; err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestWriterContextCancellationInterruptsWrite(t *testing.T) {
+	c, _ := leakCheckedPipe(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	c.BindWriteContext(ctx)
+	if err := c.Notify("queued", nil); err != nil {
+		t.Fatal(err)
+	}
+	cancel()
+	select {
+	case <-c.Done():
+	case <-time.After(time.Second):
+		t.Fatal("context cancellation did not kill stream")
+	}
+}
diff --git a/services/versions.json b/services/versions.json
index 0a255a572..b14e0eff7 100644
--- a/services/versions.json
+++ b/services/versions.json
@@ -3,19 +3,19 @@
   "product": "3.0.0",
   "installer": "3.0.0",
   "components": {
-    "ollama-proxy": "1.5.0",
-    "lmstudio-proxy": "1.5.0",
-    "nvpair-node-info": "1.4.0",
+    "ollama-proxy": "1.5.1",
+    "lmstudio-proxy": "1.5.1",
+    "nvpair-node-info": "1.4.1",
     "nvpair-node-scanner": "1.3.1",
     "nvpair-manual-nodes": "1.3.1",
-    "nvpair-workload-manager": "1.3.1",
-    "nvpair-errors": "1.1.1",
-    "nvpair-node-settings": "3.2.0",
-    "nvpair-ui-broker": "2.5.0",
-    "nvpair-engine-manager": "1.4.0",
+    "nvpair-workload-manager": "1.3.2",
+    "nvpair-errors": "1.1.2",
+    "nvpair-node-settings": "3.2.1",
+    "nvpair-ui-broker": "2.5.1",
+    "nvpair-engine-manager": "1.4.1",
     "nvpair-cluster-manager": "2.3.0",
-    "nvpair-job-scheduler": "1.1.0",
-    "nvpair-tui": "1.3.0",
-    "omlx-proxy": "1.5.0"
+    "nvpair-job-scheduler": "1.1.1",
+    "nvpair-tui": "1.3.1",
+    "omlx-proxy": "1.5.1"
   }
 }