fix(CLI-40,CLI-41,CLI-44): exact-secret scrub, uniform malformed-reply contract, Go terminal-error mislabel

CLI-40: port the exact-secret credential scrub to Rust/Java/.NET (Go/Python
already did it). AuthenticateUser/WriteSecured(2) helpers now redact the exact
caller-supplied secret from any surfaced error, as defense-in-depth on top of the
by-construction guarantee. Rust hand-writes a redacting Debug (derived Debug would
leak the reply); Java/.NET rebuild the same exception type with the redacted
message and do not carry the secret-bearing original forward (so ToString/stack
traces stay clean too).

CLI-41: uniform malformed-reply contract for AuthenticateUser/ArchestrAUserToId/
AddBufferedItem across all five clients — typed payload, else a present int32
return_value, else a typed malformed-reply error. Fixes Go/Java silent-0, .NET
NRE, and Rust's own internal inconsistency.

CLI-44: the Go event goroutine's Recv-error path now uses a non-blocking
sendTerminalEventResult on the reserved slot, so a genuine terminal stream error
is reported as itself instead of being mislabeled ErrSlowConsumer under overflow.

Riders from the CLI-37/38 review: (a) .NET ToDiagnosticSummary and Python
_mxaccess_message surface the raw success member (diagnostics-only parity with
Rust); (b) the status-conversion fixture carries an independent wantSuccess
boolean and the Go/.NET fixture tests assert against it instead of recomputing
the formula under test.

Shared fixtures (authenticate-user.{echoed-credential,missing-payload,
return-value-only}.reply.json) + manifest + ClientBehaviorFixtures.md +
ClientLibrariesDesign.md updated in the same change. Tracking: CLI-40/41/44 -> Done.

clients/rust/src/error.rs

@@ -193,17 +193,43 @@ impl std::error::Error for CommandError {}
 /// The wrapper is heap-allocated inside [`Error::MxAccess`] to keep the
 /// containing enum small. Callers can recover the reply with
 /// [`MxAccessError::reply`] or [`MxAccessError::into_reply`]. Its `Display`
-/// summarizes the `hresult` and status entries and scrubs any credential-like
-/// tokens from diagnostic text before it reaches a caller.
-#[derive(Clone, Debug)]
+/// summarizes the `hresult` and status entries and scrubs credentials from the
+/// rendered text before it reaches a caller: credential-*shaped* tokens
+/// (`mxgw_...`, `bearer`) via a pattern scrub, plus any exact caller-supplied
+/// secrets registered with [`MxAccessError::with_secrets`] — the latter catches
+/// a password MXAccess echoed back verbatim even though it has no token shape.
+///
+/// `Debug` is hand-written (not derived) so the attached exact secrets never
+/// reach `{:?}` output either: it scrubs them from the reply rendering and
+/// prints only the count of attached secrets, never their values.
+#[derive(Clone)]
 pub struct MxAccessError {
     reply: MxCommandReply,
+    /// Exact caller-supplied secrets (e.g. an `AuthenticateUser` password or a
+    /// `WriteSecured` string value) scrubbed from the rendered message. Empty
+    /// unless a helper attaches them via [`Self::with_secrets`].
+    secrets: Vec<String>,
 }
 
 impl MxAccessError {
     /// Wrap a reply whose MXAccess-level result reported a failure.
     pub fn new(reply: MxCommandReply) -> Self {
-        Self { reply }
+        Self {
+            reply,
+            secrets: Vec::new(),
+        }
+    }
+
+    /// Register exact caller-supplied secrets to scrub from the rendered
+    /// message, returning the updated error.
+    ///
+    /// A credential MXAccess echoes back into its diagnostic text has no
+    /// `mxgw_`/`bearer` shape, so the pattern scrub cannot catch it. Attaching
+    /// the exact secret lets `Display` replace every occurrence with
+    /// `<redacted>`.
+    pub fn with_secrets(mut self, secrets: Vec<String>) -> Self {
+        self.secrets = secrets;
+        self
     }
 
     /// Borrow the underlying reply (correlation id, hresult, statuses).
@@ -217,15 +243,43 @@ impl MxAccessError {
     }
 }
 
+impl std::fmt::Debug for MxAccessError {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        // Render the reply, scrub any exact caller secret from it, and never
+        // print the raw secrets themselves — only how many are attached.
+        let mut reply = format!("{:?}", self.reply);
+        for secret in &self.secrets {
+            if !secret.is_empty() {
+                reply = reply.replace(secret.as_str(), "<redacted>");
+            }
+        }
+
+        formatter
+            .debug_struct("MxAccessError")
+            .field("reply", &format_args!("{reply}"))
+            .field(
+                "secrets",
+                &format_args!("[{} redacted]", self.secrets.len()),
+            )
+            .finish()
+    }
+}
+
 impl std::fmt::Display for MxAccessError {
     fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        use std::fmt::Write as _;
+
         let hresult = match self.reply.hresult {
             Some(value) => value.to_string(),
             None => "none".to_owned(),
         };
 
+        // Render the whole body first so the exact-secret scrub can sweep every
+        // field — including diagnostic text that already went through the
+        // credential-shape scrub — before any of it reaches the caller.
+        let mut body = String::new();
         write!(
-            formatter,
+            body,
             "hresult={hresult}, {} status entr{}",
             self.reply.statuses.len(),
             if self.reply.statuses.len() == 1 {
@@ -233,20 +287,28 @@ impl std::fmt::Display for MxAccessError {
             } else {
                 "ies"
             }
-        )?;
+        )
+        .expect("writing to a String is infallible");
 
         for status in &self.reply.statuses {
             let category = MxStatusCategory::try_from(status.category)
                 .unwrap_or(MxStatusCategory::Unspecified);
             let diagnostic = redact_credentials(&status.diagnostic_text);
             write!(
-                formatter,
+                body,
                 "; [success={}, category={category:?}, detail={}, {}]",
                 status.success, status.detail, diagnostic
-            )?;
+            )
+            .expect("writing to a String is infallible");
         }
 
-        Ok(())
+        for secret in &self.secrets {
+            if !secret.is_empty() {
+                body = body.replace(secret.as_str(), "<redacted>");
+            }
+        }
+
+        formatter.write_str(&body)
     }
 }
 

clients/rust/src/session.rs

@@ -27,7 +27,7 @@ use crate::generated::mxaccess_gateway::v1::{
     WriteSecured2BulkCommand, WriteSecured2BulkEntry, WriteSecured2Command,
     WriteSecuredBulkCommand, WriteSecuredBulkEntry, WriteSecuredCommand,
 };
-use crate::value::{MxStatus, MxValue};
+use crate::value::{MxStatus, MxValue, MxValueProjection};
 
 const MAX_BULK_ITEMS: usize = 1_000;
 
@@ -801,6 +801,7 @@ impl Session {
         verifier_user_id: i32,
         value: MxValue,
     ) -> Result<(), Error> {
+        let secrets = string_secret(&value);
         self.invoke(
             MxCommandKind::WriteSecured,
             Payload::WriteSecured(WriteSecuredCommand {
@@ -811,7 +812,8 @@ impl Session {
                 value: Some(value.into_proto()),
             }),
         )
-        .await?;
+        .await
+        .map_err(|error| attach_secrets(error, secrets))?;
         Ok(())
     }
 
@@ -831,6 +833,7 @@ impl Session {
         value: MxValue,
         timestamp_value: MxValue,
     ) -> Result<(), Error> {
+        let secrets = string_secret(&value);
         self.invoke(
             MxCommandKind::WriteSecured2,
             Payload::WriteSecured2(WriteSecured2Command {
@@ -842,7 +845,8 @@ impl Session {
                 timestamp_value: Some(timestamp_value.into_proto()),
             }),
         )
-        .await?;
+        .await
+        .map_err(|error| attach_secrets(error, secrets))?;
         Ok(())
     }
 
@@ -882,7 +886,8 @@ impl Session {
                     verify_user_password: verify_user_password.to_owned(),
                 }),
             )
-            .await?;
+            .await
+            .map_err(|error| attach_secrets(error, vec![verify_user_password.to_owned()]))?;
 
         authenticate_user_id(&reply)
     }
@@ -1074,18 +1079,51 @@ fn add_buffered_item_handle(reply: &MxCommandReply) -> Result<i32, Error> {
 fn authenticate_user_id(reply: &MxCommandReply) -> Result<i32, Error> {
     match reply.payload.as_ref() {
         Some(mx_command_reply::Payload::AuthenticateUser(authenticate)) => Ok(authenticate.user_id),
-        _ => Err(Error::MalformedReply {
-            detail: "authenticate_user reply lacked an AuthenticateUser payload".to_owned(),
-        }),
+        _ => reply
+            .return_value
+            .as_ref()
+            .and_then(int32_reply_value)
+            .ok_or_else(|| Error::MalformedReply {
+                detail:
+                    "authenticate_user reply lacked an AuthenticateUser payload or int32 return_value"
+                        .to_owned(),
+            }),
     }
 }
 
 fn archestra_user_id(reply: &MxCommandReply) -> Result<i32, Error> {
     match reply.payload.as_ref() {
         Some(mx_command_reply::Payload::ArchestraUserToId(archestra)) => Ok(archestra.user_id),
-        _ => Err(Error::MalformedReply {
-            detail: "archestra_user_to_id reply lacked an ArchestraUserToId payload".to_owned(),
-        }),
+        _ => reply
+            .return_value
+            .as_ref()
+            .and_then(int32_reply_value)
+            .ok_or_else(|| Error::MalformedReply {
+                detail:
+                    "archestra_user_to_id reply lacked an ArchestraUserToId payload or int32 return_value"
+                        .to_owned(),
+            }),
+    }
+}
+
+/// Extract an exact string secret from a credential-sensitive [`MxValue`] so a
+/// failing `WriteSecured`/`WriteSecured2` can scrub it from the surfaced error.
+/// Non-string values carry no scrubbable secret and yield an empty vector.
+fn string_secret(value: &MxValue) -> Vec<String> {
+    match value.projection() {
+        MxValueProjection::String(text) if !text.is_empty() => vec![text.clone()],
+        _ => Vec::new(),
+    }
+}
+
+/// Attach caller-supplied exact secrets to an [`Error::MxAccess`] before it
+/// propagates, so its `Display` scrubs any occurrence of the credential (e.g. a
+/// password MXAccess echoed back verbatim). Any other error variant is returned
+/// unchanged.
+fn attach_secrets(error: Error, secrets: Vec<String>) -> Error {
+    match error {
+        Error::MxAccess(boxed) => Error::MxAccess(Box::new(boxed.with_secrets(secrets))),
+        other => other,
     }
 }
 

clients/rust/tests/client_behavior.rs

@@ -804,6 +804,93 @@ async fn authenticate_user_keeps_credentials_out_of_surfaced_errors() {
     );
 }
 
+#[tokio::test]
+async fn authenticate_user_scrubs_exact_caller_credential_echoed_in_diagnostic() {
+    // CLI-40: MXAccess can echo the supplied credential back inside its failure
+    // diagnostic (here in statuses[0].diagnostic_text). The token has no
+    // mxgw_/bearer shape, so the pattern scrub alone cannot catch it — the
+    // exact-secret scrub must replace the caller's password with <redacted>.
+    let credential = "sup3rSecretVerify9f3a2b";
+    let state = Arc::new(FakeState::default());
+    *state.invoke_override.lock().await = Some(InvokeOverride::CannedReply(Box::new(
+        command_reply_fixture("authenticate-user.echoed-credential.reply.json"),
+    )));
+    let endpoint = spawn_fake_gateway(state.clone()).await;
+    let client = GatewayClient::connect(ClientOptions::new(endpoint))
+        .await
+        .unwrap();
+    let session = client.session("session-fixture");
+
+    let error = session
+        .authenticate_user(7, "verifier", credential)
+        .await
+        .unwrap_err();
+
+    assert!(
+        matches!(error, Error::MxAccess(_)),
+        "OK protocol + negative hresult must route to Error::MxAccess: {error:?}"
+    );
+    let rendered = error.to_string();
+    assert!(
+        !rendered.contains(credential),
+        "exact caller credential leaked into the surfaced error: {rendered}"
+    );
+    assert!(
+        rendered.contains("<redacted>"),
+        "credential occurrence must be replaced with <redacted>: {rendered}"
+    );
+}
+
+#[tokio::test]
+async fn authenticate_user_maps_missing_payload_reply_to_malformed_reply() {
+    // CLI-41: an OK reply with neither a typed AuthenticateUser payload nor a
+    // return_value is malformed.
+    let state = Arc::new(FakeState::default());
+    *state.invoke_override.lock().await = Some(InvokeOverride::CannedReply(Box::new(
+        command_reply_fixture("authenticate-user.missing-payload.reply.json"),
+    )));
+    let endpoint = spawn_fake_gateway(state.clone()).await;
+    let client = GatewayClient::connect(ClientOptions::new(endpoint))
+        .await
+        .unwrap();
+    let session = client.session("session-fixture");
+
+    let error = session
+        .authenticate_user(7, "verifier", "pw")
+        .await
+        .unwrap_err();
+
+    assert!(
+        matches!(error, Error::MalformedReply { .. }),
+        "missing payload + missing return_value must be MalformedReply, got {error:?}"
+    );
+}
+
+#[tokio::test]
+async fn authenticate_user_falls_back_to_return_value_when_typed_payload_absent() {
+    // CLI-41: an OK reply that carries only a return_value (legacy worker) must
+    // resolve the user id from it, mirroring add_buffered_item's fallback.
+    let state = Arc::new(FakeState::default());
+    *state.invoke_override.lock().await = Some(InvokeOverride::CannedReply(Box::new(
+        command_reply_fixture("authenticate-user.return-value-only.reply.json"),
+    )));
+    let endpoint = spawn_fake_gateway(state.clone()).await;
+    let client = GatewayClient::connect(ClientOptions::new(endpoint))
+        .await
+        .unwrap();
+    let session = client.session("session-fixture");
+
+    let user_id = session
+        .authenticate_user(7, "verifier", "pw")
+        .await
+        .unwrap();
+
+    assert_eq!(
+        user_id, 7,
+        "user id must resolve from the int32 return_value"
+    );
+}
+
 #[tokio::test]
 async fn stream_alarms_emits_snapshot_then_complete_then_transition_in_order() {
     let state = Arc::new(FakeState::default());
@@ -955,6 +1042,11 @@ enum InvokeOverride {
     /// `AuthenticateUser` rejected by MXAccess) so the client's
     /// `ensure_mxaccess_success` check is exercised on the typed helper path.
     MxAccessFailure,
+    /// Reply with a caller-supplied canned [`MxCommandReply`]. Lets a test
+    /// drive a helper with a shared behavior fixture (e.g. the
+    /// echoed-credential / missing-payload / return-value-only
+    /// authenticate-user replies). Boxed to keep the enum small.
+    CannedReply(Box<MxCommandReply>),
 }
 
 #[derive(Clone)]
@@ -1057,6 +1149,7 @@ impl MxAccessGateway for FakeGateway {
                     payload: None,
                     ..MxCommandReply::default()
                 })),
+                InvokeOverride::CannedReply(reply) => Ok(Response::new(*reply)),
                 InvokeOverride::WriteOk => {
                     // Extract and capture the WriteCommand payload so the test
                     // can assert on server_handle, item_handle, user_id, and value.
@@ -1443,6 +1536,18 @@ fn command_reply_fixture(file_name: &str) -> MxCommandReply {
         })
         .collect();
 
+    // The fixtures that exercise the return_value fallback path carry a typed
+    // `returnValue` (VT_I4). Project it so a canned reply can drive the
+    // helper's payload -> return_value -> MalformedReply precedence.
+    let return_value = fixture.get("returnValue").and_then(|value| {
+        value["int32Value"].as_i64().map(|int32| MxValue {
+            data_type: MxDataType::Integer as i32,
+            variant_type: value["variantType"].as_str().unwrap_or("VT_I4").to_owned(),
+            kind: Some(Kind::Int32Value(int32 as i32)),
+            ..MxValue::default()
+        })
+    });
+
     MxCommandReply {
         session_id: fixture["sessionId"].as_str().unwrap_or_default().to_owned(),
         correlation_id: fixture["correlationId"]
@@ -1452,6 +1557,7 @@ fn command_reply_fixture(file_name: &str) -> MxCommandReply {
         protocol_status: Some(ok_status("command ok")),
         hresult: fixture["hresult"].as_i64().map(|hresult| hresult as i32),
         statuses,
+        return_value,
         ..MxCommandReply::default()
     }
 }