Bound JSON-RPC output without reordering notifications and responses

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() {

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
 }

services/lmstudio-proxy/worker_health_test.go

@@ -6,15 +6,14 @@ package main
 import (
 	"bytes"
 	"encoding/json"
-	"io"
 	"log/slog"
 	"net"
 	"strings"
 	"testing"
 )
 
 // 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) {

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)

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()
+}

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

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
 }

services/nvpair-cluster-manager/worker_health_test.go

@@ -6,15 +6,14 @@ package main
 import (
 	"bytes"
 	"encoding/json"
-	"io"
 	"log/slog"
 	"net"
 	"strings"
 	"testing"
 )
 
 // 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) {

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())

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
 }

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) {

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

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
 }