fix(CLI-40): scrub the credential from the redacted error's structured reply, route MXACCESS_FAILURE to MxAccess (Rust), fix Go Subscribe terminal-error drop

Code-review follow-up on the CLI-40/41/44 branch.

ISSUE 1 (all five, critical): the message-only scrub still leaked the
server-echoed credential through the redacted error's structured reply accessor
(.NET Reply/Statuses, Java reply()/protocolStatus(), Go MxAccessError.Reply via
errors.As, Rust reply()/into_reply(), Python raw_reply). The redacted error now
carries a scrubbed clone of the reply (protocol_status.message,
diagnostic_message, statuses[].diagnostic_text), with per-language tests asserting
the reply accessor no longer contains the credential.

ISSUE 2 (Rust, critical): ensure_command_success routed MXACCESS_FAILURE to
Error::Command (unlike the other four clients), bypassing attach_secrets and
leaking via derived Debug/Display. MXACCESS_FAILURE now routes to Error::MxAccess,
fixing the cross-client inconsistency.

ISSUE 3 (Go, important): the CLI-44 terminal send was unconditionally
non-blocking, dropping a genuine terminal error under a full buffer on the
never-drop SubscribeEvents path. It is now reserved-slot-non-blocking only for the
cancel-on-overflow path and blocking for the never-drop path.

New shared fixture authenticate-user.echoed-credential-mxaccess-failure.reply.json
wired into all five suites. Minors: whitespace-secret guard on .NET/Java redact
helpers; Java preserves exception subtype on redaction; redaction-helper unit
tests (Go/Java/.NET). Docs (ClientBehaviorFixtures.md, ClientLibrariesDesign.md)
updated to make the structured-field claim true.

clients/go/mxgateway/client_session_test.go

@@ -264,6 +264,74 @@ func TestEventsFullBufferTerminalErrorKeepsRootCause(t *testing.T) {
 	}
 }
 
+// TestSubscribeEventsFullBufferDeliversTerminalError is the CLI-44 regression for
+// the never-drop Subscribe path. SubscribeEvents/SubscribeEventsAfter use
+// cancelWhenResultBufferFull=false, so ordinary sends are blocking and uncapped and
+// can fill every slot in the results channel — including the reserved terminal slot.
+// A genuine terminal Recv error must still be delivered as the final result, never
+// silently dropped. The server sends eventBufferSize+eventBufferReservedSlots events
+// (filling every slot) and then returns a genuine gRPC error; with an unconditional
+// non-blocking terminal send the error is dropped, so this fails red until the send
+// path blocks for the never-drop mode.
+func TestSubscribeEventsFullBufferDeliversTerminalError(t *testing.T) {
+	fake := &fakeGatewayServer{
+		streamStarted:     make(chan struct{}),
+		streamDone:        make(chan struct{}),
+		streamEventCount:  eventBufferSize + eventBufferReservedSlots,
+		streamTerminalErr: status.Error(codes.Internal, "boom"),
+	}
+	client, cleanup := newBufconnClient(t, fake)
+	defer cleanup()
+	session := NewSessionForID(client, "session-1")
+
+	subscription, err := session.SubscribeEvents(context.Background())
+	if err != nil {
+		t.Fatalf("SubscribeEvents() error = %v", err)
+	}
+	defer subscription.Close()
+	<-fake.streamStarted
+
+	// Wait for the server to finish sending every event and return the terminal
+	// error, so the producer goroutine has filled every buffered slot before the
+	// terminal result is processed. That is what makes the dropped-terminal bug
+	// observable: with the buffer full, an unconditional non-blocking send discards
+	// the terminal error.
+	select {
+	case <-fake.streamDone:
+	case <-time.After(2 * time.Second):
+		t.Fatal("event stream did not stop after terminal error")
+	}
+	time.Sleep(250 * time.Millisecond)
+
+	// Drain fully. Every data event, then the terminal gRPC error as the final
+	// result, must arrive; the channel must not close without yielding it.
+	events := subscription.Events()
+	var last EventResult
+	gotResult := false
+	for {
+		select {
+		case res, ok := <-events:
+			if !ok {
+				if !gotResult {
+					t.Fatal("events channel closed without yielding any result")
+				}
+				var gwErr *GatewayError
+				if !errors.As(last.Err, &gwErr) {
+					t.Fatalf("final event result err is %T (%v), want the terminal *GatewayError; it was dropped", last.Err, last.Err)
+				}
+				if code := status.Code(last.Err); code != codes.Internal {
+					t.Fatalf("final event result gRPC code = %s, want %s", code, codes.Internal)
+				}
+				return
+			}
+			last = res
+			gotResult = true
+		case <-time.After(2 * time.Second):
+			t.Fatal("events channel did not close after terminal error")
+		}
+	}
+}
+
 func TestEventsSurfacesReplayGapSentinelAsTypedSignal(t *testing.T) {
 	fake := &fakeGatewayServer{
 		streamStarted: make(chan struct{}),

clients/go/mxgateway/command_reply_fixtures_test.go

@@ -127,3 +127,71 @@ func TestAuthenticateUserScrubsEchoedCredentialFromError(t *testing.T) {
 		t.Fatalf("surfaced error missing redaction marker: %q", message)
 	}
 }
+
+// TestAuthenticateUserScrubsEchoedCredentialFromStructuredReply is the CLI-40
+// follow-up: redacting only the rendered Error() string is not enough. The typed
+// *MxAccessError still carries the raw command reply, whose ProtocolStatus.Message,
+// DiagnosticMessage, and Statuses[].DiagnosticText echo the credential verbatim. A
+// logger dumping structured fields would reintroduce the leak, so the reply the
+// typed error carries must be a scrubbed clone. Both the OK+negative-HRESULT and the
+// MXACCESS_FAILURE fixtures route to *MxAccessError (via EnsureProtocolSuccess), so
+// both must be scrubbed identically.
+func TestAuthenticateUserScrubsEchoedCredentialFromStructuredReply(t *testing.T) {
+	const credential = "sup3rSecretVerify9f3a2b"
+	fixtures := []string{
+		"authenticate-user.echoed-credential.reply.json",
+		"authenticate-user.echoed-credential-mxaccess-failure.reply.json",
+	}
+	for _, fixture := range fixtures {
+		t.Run(fixture, func(t *testing.T) {
+			fake := &fakeGatewayServer{
+				invokeReply: loadCommandReplyFixture(t, fixture),
+			}
+			client, cleanup := newBufconnClient(t, fake)
+			defer cleanup()
+			session := NewSessionForID(client, "session-1")
+
+			_, err := session.AuthenticateUser(context.Background(), 12, "operator", credential)
+			if err == nil {
+				t.Fatal("AuthenticateUser() error = nil, want an MXAccess failure")
+			}
+
+			var mxErr *MxAccessError
+			if !errors.As(err, &mxErr) {
+				t.Fatalf("AuthenticateUser() error = %v (%T), want *MxAccessError", err, err)
+			}
+
+			reply := mxErr.Reply
+			if reply == nil {
+				t.Fatal("MxAccessError.Reply is nil, want the scrubbed command reply")
+			}
+			if got := reply.GetProtocolStatus().GetMessage(); strings.Contains(got, credential) {
+				t.Fatalf("MxAccessError.Reply.ProtocolStatus.Message leaked the credential: %q", got)
+			}
+			if got := reply.GetDiagnosticMessage(); strings.Contains(got, credential) {
+				t.Fatalf("MxAccessError.Reply.DiagnosticMessage leaked the credential: %q", got)
+			}
+			for i, status := range reply.GetStatuses() {
+				if got := status.GetDiagnosticText(); strings.Contains(got, credential) {
+					t.Fatalf("MxAccessError.Reply.Statuses[%d].DiagnosticText leaked the credential: %q", i, got)
+				}
+			}
+
+			// The wrapped CommandError's status/reply must be scrubbed too.
+			if mxErr.Command != nil {
+				if got := mxErr.Command.Status.GetMessage(); strings.Contains(got, credential) {
+					t.Fatalf("MxAccessError.Command.Status.Message leaked the credential: %q", got)
+				}
+				if cmdReply := mxErr.Command.Reply; cmdReply != nil {
+					if got := cmdReply.GetDiagnosticMessage(); strings.Contains(got, credential) {
+						t.Fatalf("MxAccessError.Command.Reply.DiagnosticMessage leaked the credential: %q", got)
+					}
+				}
+			}
+
+			if got := err.Error(); strings.Contains(got, credential) {
+				t.Fatalf("rendered error leaked the credential: %q", got)
+			}
+		})
+	}
+}

clients/go/mxgateway/errors.go

@@ -6,6 +6,7 @@ import (
 	"strings"
 
 	pb "gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/internal/generated"
+	"google.golang.org/protobuf/proto"
 )
 
 // redactedSecretMarker is the placeholder substituted for credential material in
@@ -49,20 +50,108 @@ func (e *secretRedactingError) Unwrap() error {
 	return e.err
 }
 
-// redactSecrets wraps err so any occurrence of a non-empty secret in the surfaced
-// message is redacted, while errors.As / errors.Is still reach the wrapped typed
-// error. It returns nil unchanged and skips wrapping when no non-empty secret is
-// supplied, so non-secret-bearing calls keep their original error verbatim.
+// scrubReplyStrings returns a clone of reply with every non-empty secret replaced
+// by redactedSecretMarker in the free-text fields a gateway diagnostic could echo a
+// credential into: ProtocolStatus.Message, DiagnosticMessage, and each
+// Statuses[].DiagnosticText. It clones with proto.Clone so the caller's original
+// reply is never mutated. A nil reply, or an empty/whitespace-only secret set, is a
+// no-op (nil in, nil out; a clone otherwise).
+func scrubReplyStrings(reply *pb.MxCommandReply, secrets []string) *pb.MxCommandReply {
+	if reply == nil {
+		return nil
+	}
+	clone, ok := proto.Clone(reply).(*pb.MxCommandReply)
+	if !ok {
+		return reply
+	}
+	for _, secret := range secrets {
+		if secret == "" {
+			continue
+		}
+		if clone.GetProtocolStatus() != nil {
+			clone.ProtocolStatus.Message = strings.ReplaceAll(clone.GetProtocolStatus().GetMessage(), secret, redactedSecretMarker)
+		}
+		clone.DiagnosticMessage = strings.ReplaceAll(clone.GetDiagnosticMessage(), secret, redactedSecretMarker)
+		for _, status := range clone.GetStatuses() {
+			status.DiagnosticText = strings.ReplaceAll(status.GetDiagnosticText(), secret, redactedSecretMarker)
+		}
+	}
+	return clone
+}
+
+// scrubProtocolStatusMessage returns a clone of status with every non-empty secret
+// redacted from its Message, leaving the original untouched.
+func scrubProtocolStatusMessage(status *ProtocolStatus, secrets []string) *ProtocolStatus {
+	if status == nil {
+		return nil
+	}
+	clone, ok := proto.Clone(status).(*ProtocolStatus)
+	if !ok {
+		return status
+	}
+	for _, secret := range secrets {
+		if secret != "" {
+			clone.Message = strings.ReplaceAll(clone.GetMessage(), secret, redactedSecretMarker)
+		}
+	}
+	return clone
+}
+
+// redactSecrets scrubs a non-empty secret set from the error it surfaces. When the
+// wrapped error is a typed *MxAccessError or *CommandError it is rebuilt carrying
+// scrubbed clones of its reply and protocol status, so a caller logging the typed
+// error's structured fields cannot reintroduce the credential the rendered message
+// hides. The rebuilt (or original, for other error types) value is then wrapped in
+// secretRedactingError as a belt-and-suspenders scrub of any remaining rendered
+// text. errors.As / errors.Is still reach the typed error through the wrapper. It
+// returns nil unchanged and skips all work when no non-empty secret is supplied, so
+// non-secret-bearing calls keep their original error verbatim.
 func redactSecrets(err error, secrets ...string) error {
 	if err == nil {
 		return nil
 	}
+	hasSecret := false
 	for _, secret := range secrets {
 		if secret != "" {
-			return &secretRedactingError{err: err, secrets: secrets}
+			hasSecret = true
+			break
 		}
 	}
-	return err
+	if !hasSecret {
+		return err
+	}
+
+	rebuilt := rebuildScrubbedError(err, secrets)
+	return &secretRedactingError{err: rebuilt, secrets: secrets}
+}
+
+// rebuildScrubbedError rebuilds the typed error carrying scrubbed clones of any
+// command reply / protocol status it holds, so credential text never survives in the
+// error's structured fields. Non-reply-bearing error types are returned unchanged.
+func rebuildScrubbedError(err error, secrets []string) error {
+	switch typed := err.(type) {
+	case *MxAccessError:
+		return &MxAccessError{
+			Command: scrubCommandError(typed.Command, secrets),
+			Reply:   scrubReplyStrings(typed.Reply, secrets),
+		}
+	case *CommandError:
+		return scrubCommandError(typed, secrets)
+	default:
+		return err
+	}
+}
+
+// scrubCommandError rebuilds a CommandError with a scrubbed Status and Reply.
+func scrubCommandError(cmd *CommandError, secrets []string) *CommandError {
+	if cmd == nil {
+		return nil
+	}
+	return &CommandError{
+		Op:     cmd.Op,
+		Status: scrubProtocolStatusMessage(cmd.Status, secrets),
+		Reply:  scrubReplyStrings(cmd.Reply, secrets),
+	}
 }
 
 // ErrSlowConsumer is the terminal error sent on the Events/EventsAfter

clients/go/mxgateway/errors_redaction_test.go

@@ -0,0 +1,115 @@
+package mxgateway
+
+import (
+	"errors"
+	"strings"
+	"testing"
+
+	pb "gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/internal/generated"
+)
+
+// TestScrubReplyStringsRedactsEveryOccurrence covers the multi-occurrence case:
+// one secret appearing across ProtocolStatus.Message, DiagnosticMessage, and every
+// Statuses[].DiagnosticText must be fully redacted with no residue.
+func TestScrubReplyStringsRedactsEveryOccurrence(t *testing.T) {
+	const secret = "hunter2"
+	reply := &pb.MxCommandReply{
+		ProtocolStatus:    &pb.ProtocolStatus{Message: "rejected hunter2 then hunter2 again"},
+		DiagnosticMessage: "echoed hunter2 back",
+		Statuses: []*pb.MxStatusProxy{
+			{DiagnosticText: "first hunter2"},
+			{DiagnosticText: "second hunter2 and hunter2"},
+		},
+	}
+
+	scrubbed := scrubReplyStrings(reply, []string{secret})
+
+	if strings.Contains(scrubbed.GetProtocolStatus().GetMessage(), secret) {
+		t.Fatalf("ProtocolStatus.Message still contains the secret: %q", scrubbed.GetProtocolStatus().GetMessage())
+	}
+	if strings.Contains(scrubbed.GetDiagnosticMessage(), secret) {
+		t.Fatalf("DiagnosticMessage still contains the secret: %q", scrubbed.GetDiagnosticMessage())
+	}
+	for i, status := range scrubbed.GetStatuses() {
+		if strings.Contains(status.GetDiagnosticText(), secret) {
+			t.Fatalf("Statuses[%d].DiagnosticText still contains the secret: %q", i, status.GetDiagnosticText())
+		}
+	}
+	if !strings.Contains(scrubbed.GetProtocolStatus().GetMessage(), redactedSecretMarker) {
+		t.Fatalf("ProtocolStatus.Message missing redaction marker: %q", scrubbed.GetProtocolStatus().GetMessage())
+	}
+
+	// The original reply must be untouched (scrubReplyStrings clones).
+	if !strings.Contains(reply.GetDiagnosticMessage(), secret) {
+		t.Fatal("scrubReplyStrings mutated the original reply instead of cloning it")
+	}
+}
+
+// TestScrubReplyStringsRedactsOverlappingSecrets covers two secrets where one is a
+// substring of the other: both must be fully redacted, with no partial leak of the
+// longer secret's non-shared remainder.
+func TestScrubReplyStringsRedactsOverlappingSecrets(t *testing.T) {
+	const shortSecret = "pass"
+	const longSecret = "password123"
+	reply := &pb.MxCommandReply{
+		DiagnosticMessage: "value was password123 and also pass",
+	}
+
+	scrubbed := scrubReplyStrings(reply, []string{longSecret, shortSecret})
+
+	got := scrubbed.GetDiagnosticMessage()
+	if strings.Contains(got, shortSecret) {
+		t.Fatalf("scrubbed message still contains a secret substring %q: %q", shortSecret, got)
+	}
+	if strings.Contains(got, longSecret) {
+		t.Fatalf("scrubbed message still contains %q: %q", longSecret, got)
+	}
+	// "123" is the longer secret's remainder past the shared "pass" prefix; it must
+	// not survive as a partial leak.
+	if strings.Contains(got, "123") {
+		t.Fatalf("scrubbed message leaked the longer secret's remainder: %q", got)
+	}
+}
+
+// TestRedactSecretsEmptyOrNilLeavesErrorUnchanged confirms the no-secret paths keep
+// the original typed error verbatim (no wrapping, no scrubbed clone).
+func TestRedactSecretsEmptyOrNilLeavesErrorUnchanged(t *testing.T) {
+	base := &MxAccessError{Reply: &pb.MxCommandReply{DiagnosticMessage: "boom"}}
+
+	if got := redactSecrets(base); got != error(base) {
+		t.Fatalf("redactSecrets with no secrets = %v, want the original error unchanged", got)
+	}
+	if got := redactSecrets(base, ""); got != error(base) {
+		t.Fatalf("redactSecrets with only an empty secret = %v, want the original error unchanged", got)
+	}
+	if got := redactSecrets(nil, "secret"); got != nil {
+		t.Fatalf("redactSecrets(nil, ...) = %v, want nil", got)
+	}
+}
+
+// TestRedactSecretsRebuildsTypedCommandError confirms a *CommandError (non-MXAccess
+// path) is rebuilt with a scrubbed Status and Reply, and errors.As still reaches it.
+func TestRedactSecretsRebuildsTypedCommandError(t *testing.T) {
+	const secret = "topSecretValue"
+	base := &CommandError{
+		Op:     "write secured",
+		Status: &pb.ProtocolStatus{Message: "rejected topSecretValue"},
+		Reply:  &pb.MxCommandReply{DiagnosticMessage: "echoed topSecretValue"},
+	}
+
+	redacted := redactSecrets(base, secret)
+
+	var cmdErr *CommandError
+	if !errors.As(redacted, &cmdErr) {
+		t.Fatalf("redactSecrets result %T does not unwrap to *CommandError", redacted)
+	}
+	if strings.Contains(cmdErr.Status.GetMessage(), secret) {
+		t.Fatalf("CommandError.Status.Message leaked the secret: %q", cmdErr.Status.GetMessage())
+	}
+	if strings.Contains(cmdErr.Reply.GetDiagnosticMessage(), secret) {
+		t.Fatalf("CommandError.Reply.DiagnosticMessage leaked the secret: %q", cmdErr.Reply.GetDiagnosticMessage())
+	}
+	if strings.Contains(redacted.Error(), secret) {
+		t.Fatalf("rendered error leaked the secret: %q", redacted.Error())
+	}
+}

clients/go/mxgateway/session.go

@@ -1073,8 +1073,8 @@ func (s *Session) subscribeEventsAfter(ctx context.Context, afterWorkerSequence
 			// A genuine terminal stream error must be reported as itself, even
 			// when the data slots are full. Routing it through sendEventResult
 			// would let the overflow branch substitute ErrSlowConsumer and lose
-			// the real gRPC status, so send it directly on the reserved slot.
-			sendTerminalEventResult(results, EventResult{Err: &GatewayError{Op: "stream events", Err: err}})
+			// the real gRPC status, so send it directly, bypassing that branch.
+			sendTerminalEventResult(streamCtx, results, EventResult{Err: &GatewayError{Op: "stream events", Err: err}}, cancelWhenResultBufferFull)
 			return
 		}
 	}()
@@ -1093,20 +1093,32 @@ func ensureBulkSize(name string, length int) error {
 	return nil
 }
 
-// sendTerminalEventResult enqueues a terminal EventResult with a non-blocking
-// send. The eventBufferReservedSlots reserve (beyond the eventBufferSize data
-// slots) guarantees the send lands unless a terminal result was already
-// enqueued; because this goroutine is the sole producer, at most one terminal
-// send ever races for the reserved slot, so the select default only fires when
-// the reserve is already spent — never dropping a first terminal error.
+// sendTerminalEventResult enqueues a terminal EventResult, bypassing
+// sendEventResult's overflow branch so a genuine stream error is reported verbatim
+// rather than relabeled as ErrSlowConsumer. How it sends depends on the mode:
 //
-// Unlike sendEventResult, this bypasses the overflow branch: a genuine terminal
-// stream error is reported verbatim even when the data slots are full, rather
-// than being relabeled as ErrSlowConsumer.
-func sendTerminalEventResult(results chan<- EventResult, result EventResult) {
+//   - cancelWhenBufferFull=true (Events/EventsAfter): ordinary data sends are capped
+//     at eventBufferSize, leaving eventBufferReservedSlots free, so a non-blocking
+//     send always lands the terminal result. Because this goroutine is the sole
+//     producer, at most one terminal send ever races for the reserved slot, so the
+//     select default only fires when the reserve is already spent — never dropping a
+//     first terminal error.
+//   - cancelWhenBufferFull=false (SubscribeEvents/SubscribeEventsAfter, never-drop):
+//     ordinary data sends are uncapped and blocking, so every slot including the
+//     reserve can hold data. A non-blocking send would then hit the full buffer and
+//     silently drop the terminal error, breaking the never-drop contract; instead
+//     block until the consumer drains a slot (or the stream context is cancelled).
+func sendTerminalEventResult(ctx context.Context, results chan<- EventResult, result EventResult, cancelWhenBufferFull bool) {
+	if cancelWhenBufferFull {
+		select {
+		case results <- result:
+		default:
+		}
+		return
+	}
 	select {
 	case results <- result:
-	default:
+	case <-ctx.Done():
 	}
 }