Add bulk read/write CLI subcommands and e2e matrix coverage

The previous commit added the bulk read/write library surface in every
client; this commit makes that surface reachable from each client's CLI
and exercises it through scripts/run-client-e2e-tests.ps1.

Five new subcommands in every client CLI (.NET / Go / Rust / Python /
Java): read-bulk, write-bulk, write2-bulk, write-secured-bulk, and
write-secured2-bulk. Each follows the existing subscribe-bulk shape:

  - read-bulk takes --server-handle, --items <csv tag list>, and
    --timeout-ms (0 = worker default). JSON output carries the
    BulkReadResult fields, including was_cached so the e2e matrix can
    verify the cached-path semantics.
  - The four bulk-write families take --server-handle, --item-handles
    <csv>, --type, --values <csv>. write2-bulk and write-secured2-bulk
    add a single --timestamp applied to every entry; the secured
    variants take --current-user-id and --verifier-user-id. All four
    output BulkWriteResult JSON.

A new -SkipReadWriteBulk switch on the matrix script (default OFF)
controls two new e2e phases:

  - After the existing subscribe-bulk phase leaves tags advised, the
    script runs read-bulk against the same tag list and asserts most
    results return was_cached = true. This is the only e2e coverage of
    the cache-then-snapshot fork — the unit + gateway tests verify the
    semantics with a fake worker, but only the live cross-language
    matrix proves the cache populates from real OnDataChange events and
    survives the round-trip through every client''s JSON parser.
  - When -VerifyWrite is set, the write phase now also runs a single-
    entry write-bulk against the same writable item handle (using a
    distinct sentinel value) and asserts a per-entry success. Confirms
    the BulkWriteResult wire format end-to-end without complicating
    the OnWriteComplete echo assertion the single-item phase already
    verifies.

Dry-run validation passes for all five clients: each emits the correct
read-bulk and write-bulk CLI invocations with the right flags.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

clients/dotnet/MxGateway.Client.Cli/MxGatewayClientCli.cs

@@ -101,6 +101,16 @@ public static class MxGatewayClientCli
                     .ConfigureAwait(false),
                 "unsubscribe-bulk" => await UnsubscribeBulkAsync(arguments, client, standardOutput, cancellation.Token)
                     .ConfigureAwait(false),
+                "read-bulk" => await ReadBulkAsync(arguments, client, standardOutput, cancellation.Token)
+                    .ConfigureAwait(false),
+                "write-bulk" => await WriteBulkAsync(arguments, client, standardOutput, cancellation.Token)
+                    .ConfigureAwait(false),
+                "write2-bulk" => await Write2BulkAsync(arguments, client, standardOutput, cancellation.Token)
+                    .ConfigureAwait(false),
+                "write-secured-bulk" => await WriteSecuredBulkAsync(arguments, client, standardOutput, cancellation.Token)
+                    .ConfigureAwait(false),
+                "write-secured2-bulk" => await WriteSecured2BulkAsync(arguments, client, standardOutput, cancellation.Token)
+                    .ConfigureAwait(false),
                 "stream-events" => await StreamEventsAsync(arguments, client, standardOutput, cancellation.Token)
                     .ConfigureAwait(false),
                 "write" => await WriteAsync(arguments, client, standardOutput, cancellation.Token)
@@ -386,6 +396,220 @@ public static class MxGatewayClientCli
             cancellationToken);
     }
 
+    private static Task<int> ReadBulkAsync(
+        CliArguments arguments,
+        IMxGatewayCliClient client,
+        TextWriter output,
+        CancellationToken cancellationToken)
+    {
+        ReadBulkCommand command = new()
+        {
+            ServerHandle = arguments.GetInt32("server-handle"),
+            TimeoutMs = (uint)arguments.GetInt32("timeout-ms", 0),
+        };
+        command.TagAddresses.Add(ParseStringList(arguments.GetRequired("items")));
+
+        return InvokeAndWriteAsync(
+            arguments,
+            client,
+            output,
+            new MxCommand
+            {
+                Kind = MxCommandKind.ReadBulk,
+                ReadBulk = command,
+            },
+            cancellationToken);
+    }
+
+    private static Task<int> WriteBulkAsync(
+        CliArguments arguments,
+        IMxGatewayCliClient client,
+        TextWriter output,
+        CancellationToken cancellationToken)
+    {
+        WriteBulkCommand command = new()
+        {
+            ServerHandle = arguments.GetInt32("server-handle"),
+        };
+
+        IReadOnlyList<int> handles = ParseInt32List(arguments.GetRequired("item-handles"));
+        IReadOnlyList<MxValue> values = ParseValuesList(arguments);
+        int userId = arguments.GetInt32("user-id", 0);
+        EnsureSameLength(handles.Count, values.Count);
+
+        for (int i = 0; i < handles.Count; i++)
+        {
+            command.Entries.Add(new WriteBulkEntry
+            {
+                ItemHandle = handles[i],
+                Value = values[i],
+                UserId = userId,
+            });
+        }
+
+        return InvokeAndWriteAsync(
+            arguments,
+            client,
+            output,
+            new MxCommand
+            {
+                Kind = MxCommandKind.WriteBulk,
+                WriteBulk = command,
+            },
+            cancellationToken);
+    }
+
+    private static Task<int> Write2BulkAsync(
+        CliArguments arguments,
+        IMxGatewayCliClient client,
+        TextWriter output,
+        CancellationToken cancellationToken)
+    {
+        Write2BulkCommand command = new()
+        {
+            ServerHandle = arguments.GetInt32("server-handle"),
+        };
+
+        IReadOnlyList<int> handles = ParseInt32List(arguments.GetRequired("item-handles"));
+        IReadOnlyList<MxValue> values = ParseValuesList(arguments);
+        MxValue timestampValue = ParseTimestampValue(arguments);
+        int userId = arguments.GetInt32("user-id", 0);
+        EnsureSameLength(handles.Count, values.Count);
+
+        for (int i = 0; i < handles.Count; i++)
+        {
+            command.Entries.Add(new Write2BulkEntry
+            {
+                ItemHandle = handles[i],
+                Value = values[i],
+                TimestampValue = timestampValue,
+                UserId = userId,
+            });
+        }
+
+        return InvokeAndWriteAsync(
+            arguments,
+            client,
+            output,
+            new MxCommand
+            {
+                Kind = MxCommandKind.Write2Bulk,
+                Write2Bulk = command,
+            },
+            cancellationToken);
+    }
+
+    private static Task<int> WriteSecuredBulkAsync(
+        CliArguments arguments,
+        IMxGatewayCliClient client,
+        TextWriter output,
+        CancellationToken cancellationToken)
+    {
+        WriteSecuredBulkCommand command = new()
+        {
+            ServerHandle = arguments.GetInt32("server-handle"),
+        };
+
+        IReadOnlyList<int> handles = ParseInt32List(arguments.GetRequired("item-handles"));
+        IReadOnlyList<MxValue> values = ParseValuesList(arguments);
+        int currentUserId = arguments.GetInt32("current-user-id");
+        int verifierUserId = arguments.GetInt32("verifier-user-id", 0);
+        EnsureSameLength(handles.Count, values.Count);
+
+        for (int i = 0; i < handles.Count; i++)
+        {
+            command.Entries.Add(new WriteSecuredBulkEntry
+            {
+                ItemHandle = handles[i],
+                Value = values[i],
+                CurrentUserId = currentUserId,
+                VerifierUserId = verifierUserId,
+            });
+        }
+
+        return InvokeAndWriteAsync(
+            arguments,
+            client,
+            output,
+            new MxCommand
+            {
+                Kind = MxCommandKind.WriteSecuredBulk,
+                WriteSecuredBulk = command,
+            },
+            cancellationToken);
+    }
+
+    private static Task<int> WriteSecured2BulkAsync(
+        CliArguments arguments,
+        IMxGatewayCliClient client,
+        TextWriter output,
+        CancellationToken cancellationToken)
+    {
+        WriteSecured2BulkCommand command = new()
+        {
+            ServerHandle = arguments.GetInt32("server-handle"),
+        };
+
+        IReadOnlyList<int> handles = ParseInt32List(arguments.GetRequired("item-handles"));
+        IReadOnlyList<MxValue> values = ParseValuesList(arguments);
+        MxValue timestampValue = ParseTimestampValue(arguments);
+        int currentUserId = arguments.GetInt32("current-user-id");
+        int verifierUserId = arguments.GetInt32("verifier-user-id", 0);
+        EnsureSameLength(handles.Count, values.Count);
+
+        for (int i = 0; i < handles.Count; i++)
+        {
+            command.Entries.Add(new WriteSecured2BulkEntry
+            {
+                ItemHandle = handles[i],
+                Value = values[i],
+                TimestampValue = timestampValue,
+                CurrentUserId = currentUserId,
+                VerifierUserId = verifierUserId,
+            });
+        }
+
+        return InvokeAndWriteAsync(
+            arguments,
+            client,
+            output,
+            new MxCommand
+            {
+                Kind = MxCommandKind.WriteSecured2Bulk,
+                WriteSecured2Bulk = command,
+            },
+            cancellationToken);
+    }
+
+    /// <summary>
+    ///     Parses the bulk-write CLI's <c>--values</c> list. All entries share
+    ///     the single <c>--type</c> argument; the comma-separated values are
+    ///     each parsed via <see cref="ParseValue"/> on a per-entry basis. This
+    ///     keeps the CLI simple for e2e use (one type, N values) — callers
+    ///     that need heterogeneous types per entry should drive the library
+    ///     directly.
+    /// </summary>
+    private static IReadOnlyList<MxValue> ParseValuesList(CliArguments arguments)
+    {
+        string type = arguments.GetRequired("type");
+        string[] values = ParseStringList(arguments.GetRequired("values")).ToArray();
+        MxValue[] result = new MxValue[values.Length];
+        for (int i = 0; i < values.Length; i++)
+        {
+            result[i] = ParseValue(type, values[i]);
+        }
+        return result;
+    }
+
+    private static void EnsureSameLength(int handles, int values)
+    {
+        if (handles != values)
+        {
+            throw new ArgumentException(
+                $"Bulk write requires the same number of --item-handles ({handles}) and --values ({values}).");
+        }
+    }
+
     private static Task<int> WriteAsync(
         CliArguments arguments,
         IMxGatewayCliClient client,
@@ -772,8 +996,12 @@ public static class MxGatewayClientCli
 
     private static MxValue ParseValue(CliArguments arguments)
     {
-        string type = arguments.GetRequired("type").ToLowerInvariant();
+        return ParseValue(arguments.GetRequired("type"), arguments.GetRequired("value"));
-        string value = arguments.GetRequired("value");
+    }
+
+    private static MxValue ParseValue(string typeName, string value)
+    {
+        string type = typeName.ToLowerInvariant();
         string[] values = value.Split(',', StringSplitOptions.TrimEntries);
 
         return type switch

clients/go/cmd/mxgw-go/main.go

@@ -89,6 +89,16 @@ func runWithIO(ctx context.Context, args []string, stdout, stderr io.Writer) err
 		return runSubscribeBulk(ctx, args[1:], stdout, stderr)
 	case "unsubscribe-bulk":
 		return runUnsubscribeBulk(ctx, args[1:], stdout, stderr)
+	case "read-bulk":
+		return runReadBulk(ctx, args[1:], stdout, stderr)
+	case "write-bulk":
+		return runWriteBulk(ctx, args[1:], stdout, stderr)
+	case "write2-bulk":
+		return runWrite2Bulk(ctx, args[1:], stdout, stderr)
+	case "write-secured-bulk":
+		return runWriteSecuredBulk(ctx, args[1:], stdout, stderr)
+	case "write-secured2-bulk":
+		return runWriteSecured2Bulk(ctx, args[1:], stdout, stderr)
 	case "write":
 		return runWrite(ctx, args[1:], stdout, stderr)
 	case "stream-events":
@@ -347,6 +357,167 @@ func runUnsubscribeBulk(ctx context.Context, args []string, stdout, stderr io.Wr
 	return writeBulkOutput(stdout, *jsonOutput, "unsubscribe-bulk", options, results, err)
 }
 
+func runReadBulk(ctx context.Context, args []string, stdout, stderr io.Writer) error {
+	flags := flag.NewFlagSet("read-bulk", flag.ContinueOnError)
+	flags.SetOutput(stderr)
+	common := bindCommonFlags(flags)
+	jsonOutput := flags.Bool("json", false, "write JSON output")
+	sessionID := flags.String("session-id", "", "gateway session id")
+	serverHandle := flags.Int("server-handle", 0, "MXAccess server handle")
+	items := flags.String("items", "", "comma-separated tag addresses")
+	timeoutMs := flags.Int("timeout-ms", 0, "per-tag snapshot timeout in milliseconds (0 = worker default)")
+
+	if err := flags.Parse(args); err != nil {
+		return err
+	}
+	if *sessionID == "" || *items == "" {
+		return errors.New("session-id and items are required")
+	}
+
+	client, options, err := dialForCommand(ctx, common)
+	if err != nil {
+		return err
+	}
+	defer client.Close()
+
+	session := mxgateway.NewSessionForID(client, *sessionID)
+	results, err := session.ReadBulk(ctx, int32(*serverHandle), parseStringList(*items), time.Duration(*timeoutMs)*time.Millisecond)
+	return writeReadBulkOutput(stdout, *jsonOutput, "read-bulk", options, results, err)
+}
+
+func runWriteBulk(ctx context.Context, args []string, stdout, stderr io.Writer) error {
+	return runWriteBulkVariant(ctx, args, stdout, stderr, "write-bulk", false, false)
+}
+
+func runWrite2Bulk(ctx context.Context, args []string, stdout, stderr io.Writer) error {
+	return runWriteBulkVariant(ctx, args, stdout, stderr, "write2-bulk", true, false)
+}
+
+func runWriteSecuredBulk(ctx context.Context, args []string, stdout, stderr io.Writer) error {
+	return runWriteBulkVariant(ctx, args, stdout, stderr, "write-secured-bulk", false, true)
+}
+
+func runWriteSecured2Bulk(ctx context.Context, args []string, stdout, stderr io.Writer) error {
+	return runWriteBulkVariant(ctx, args, stdout, stderr, "write-secured2-bulk", true, true)
+}
+
+// runWriteBulkVariant shares the flag-parsing + entry-build skeleton across
+// the four bulk-write families. withTimestamp adds a --timestamp-value flag;
+// secured switches from --user-id to --current-user-id / --verifier-user-id.
+func runWriteBulkVariant(ctx context.Context, args []string, stdout, stderr io.Writer, command string, withTimestamp bool, secured bool) error {
+	flags := flag.NewFlagSet(command, flag.ContinueOnError)
+	flags.SetOutput(stderr)
+	common := bindCommonFlags(flags)
+	jsonOutput := flags.Bool("json", false, "write JSON output")
+	sessionID := flags.String("session-id", "", "gateway session id")
+	serverHandle := flags.Int("server-handle", 0, "MXAccess server handle")
+	itemHandles := flags.String("item-handles", "", "comma-separated item handles")
+	valueType := flags.String("type", "string", "value type: bool, int32, int64, float, double, string")
+	values := flags.String("values", "", "comma-separated values (one per item handle)")
+	userID := flags.Int("user-id", 0, "MXAccess user id (Write/Write2 variants)")
+	currentUserID := flags.Int("current-user-id", 0, "MXAccess current user id (Secured variants)")
+	verifierUserID := flags.Int("verifier-user-id", 0, "MXAccess verifier user id (Secured variants)")
+	timestampValue := flags.String("timestamp-value", "", "RFC 3339 timestamp shared across all entries (Write2/WriteSecured2 variants)")
+
+	if err := flags.Parse(args); err != nil {
+		return err
+	}
+	if *sessionID == "" || *itemHandles == "" || *values == "" {
+		return errors.New("session-id, item-handles, and values are required")
+	}
+
+	handles, err := parseInt32List(*itemHandles)
+	if err != nil {
+		return err
+	}
+	valueTexts := parseStringList(*values)
+	if len(handles) != len(valueTexts) {
+		return fmt.Errorf("item-handles count (%d) does not match values count (%d)", len(handles), len(valueTexts))
+	}
+
+	parsedValues := make([]*mxgateway.MxValue, len(handles))
+	for i, text := range valueTexts {
+		v, err := parseValue(*valueType, text)
+		if err != nil {
+			return fmt.Errorf("entry %d: %w", i, err)
+		}
+		parsedValues[i] = v
+	}
+
+	var tsValue *mxgateway.MxValue
+	if withTimestamp {
+		if *timestampValue == "" {
+			return errors.New("timestamp-value is required for write2/write-secured2 bulk variants")
+		}
+		parsed, err := parseRfc3339Timestamp(*timestampValue)
+		if err != nil {
+			return err
+		}
+		tsValue = parsed
+	}
+
+	client, options, err := dialForCommand(ctx, common)
+	if err != nil {
+		return err
+	}
+	defer client.Close()
+
+	session := mxgateway.NewSessionForID(client, *sessionID)
+
+	var results []*mxgateway.BulkWriteResult
+	switch command {
+	case "write-bulk":
+		entries := make([]*mxgateway.WriteBulkEntry, len(handles))
+		for i := range handles {
+			entries[i] = &mxgateway.WriteBulkEntry{ItemHandle: handles[i], Value: parsedValues[i], UserId: int32(*userID)}
+		}
+		results, err = session.WriteBulk(ctx, int32(*serverHandle), entries)
+	case "write2-bulk":
+		entries := make([]*mxgateway.Write2BulkEntry, len(handles))
+		for i := range handles {
+			entries[i] = &mxgateway.Write2BulkEntry{ItemHandle: handles[i], Value: parsedValues[i], TimestampValue: tsValue, UserId: int32(*userID)}
+		}
+		results, err = session.Write2Bulk(ctx, int32(*serverHandle), entries)
+	case "write-secured-bulk":
+		entries := make([]*mxgateway.WriteSecuredBulkEntry, len(handles))
+		for i := range handles {
+			entries[i] = &mxgateway.WriteSecuredBulkEntry{
+				ItemHandle:     handles[i],
+				Value:          parsedValues[i],
+				CurrentUserId:  int32(*currentUserID),
+				VerifierUserId: int32(*verifierUserID),
+			}
+		}
+		results, err = session.WriteSecuredBulk(ctx, int32(*serverHandle), entries)
+	case "write-secured2-bulk":
+		entries := make([]*mxgateway.WriteSecured2BulkEntry, len(handles))
+		for i := range handles {
+			entries[i] = &mxgateway.WriteSecured2BulkEntry{
+				ItemHandle:     handles[i],
+				Value:          parsedValues[i],
+				TimestampValue: tsValue,
+				CurrentUserId:  int32(*currentUserID),
+				VerifierUserId: int32(*verifierUserID),
+			}
+		}
+		results, err = session.WriteSecured2Bulk(ctx, int32(*serverHandle), entries)
+	default:
+		return fmt.Errorf("unsupported bulk write command %q", command)
+	}
+	_ = secured // currently only used for routing above; reserved for future per-variant validation
+	return writeWriteBulkOutput(stdout, *jsonOutput, command, options, results, err)
+}
+
+// parseRfc3339Timestamp parses an RFC 3339 timestamp and returns the
+// MxValue protobuf representation used for the timestamped write families.
+func parseRfc3339Timestamp(text string) (*mxgateway.MxValue, error) {
+	t, err := time.Parse(time.RFC3339Nano, text)
+	if err != nil {
+		return nil, fmt.Errorf("invalid RFC 3339 timestamp %q: %w", text, err)
+	}
+	return mxgateway.TimestampValue(t), nil
+}
+
 func runWrite(ctx context.Context, args []string, stdout, stderr io.Writer) error {
 	flags := flag.NewFlagSet("write", flag.ContinueOnError)
 	flags.SetOutput(stderr)
@@ -652,6 +823,36 @@ func writeBulkOutput(stdout io.Writer, jsonOutput bool, command string, options
 	return nil
 }
 
+func writeWriteBulkOutput(stdout io.Writer, jsonOutput bool, command string, options commonOptions, results []*mxgateway.BulkWriteResult, err error) error {
+	if err != nil {
+		return err
+	}
+	if jsonOutput {
+		return writeJSON(stdout, map[string]any{
+			"command": command,
+			"options": options,
+			"results": results,
+		})
+	}
+	fmt.Fprintln(stdout, len(results))
+	return nil
+}
+
+func writeReadBulkOutput(stdout io.Writer, jsonOutput bool, command string, options commonOptions, results []*mxgateway.BulkReadResult, err error) error {
+	if err != nil {
+		return err
+	}
+	if jsonOutput {
+		return writeJSON(stdout, map[string]any{
+			"command": command,
+			"options": options,
+			"results": results,
+		})
+	}
+	fmt.Fprintln(stdout, len(results))
+	return nil
+}
+
 func writeJSON(writer io.Writer, value any) error {
 	encoder := json.NewEncoder(writer)
 	encoder.SetIndent("", "  ")

clients/java/mxgateway-cli/src/main/java/com/dohertylan/mxgateway/cli/MxGatewayCli.java

@@ -25,12 +25,18 @@ import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.Callable;
+import mxaccess_gateway.v1.MxaccessGateway.BulkReadResult;
+import mxaccess_gateway.v1.MxaccessGateway.BulkWriteResult;
 import mxaccess_gateway.v1.MxaccessGateway.CloseSessionRequest;
 import mxaccess_gateway.v1.MxaccessGateway.MxCommandReply;
 import mxaccess_gateway.v1.MxaccessGateway.MxEvent;
 import mxaccess_gateway.v1.MxaccessGateway.MxValue;
 import mxaccess_gateway.v1.MxaccessGateway.OpenSessionRequest;
 import mxaccess_gateway.v1.MxaccessGateway.SubscribeResult;
+import mxaccess_gateway.v1.MxaccessGateway.Write2BulkEntry;
+import mxaccess_gateway.v1.MxaccessGateway.WriteBulkEntry;
+import mxaccess_gateway.v1.MxaccessGateway.WriteSecured2BulkEntry;
+import mxaccess_gateway.v1.MxaccessGateway.WriteSecuredBulkEntry;
 import picocli.CommandLine;
 import picocli.CommandLine.Command;
 import picocli.CommandLine.Mixin;
@@ -109,6 +115,11 @@ public final class MxGatewayCli implements Callable<Integer> {
         commandLine.addSubcommand("advise", new AdviseCommand(clientFactory));
         commandLine.addSubcommand("subscribe-bulk", new SubscribeBulkCommand(clientFactory));
         commandLine.addSubcommand("unsubscribe-bulk", new UnsubscribeBulkCommand(clientFactory));
+        commandLine.addSubcommand("read-bulk", new ReadBulkCommand(clientFactory));
+        commandLine.addSubcommand("write-bulk", new WriteBulkCommand(clientFactory));
+        commandLine.addSubcommand("write2-bulk", new Write2BulkCommand(clientFactory));
+        commandLine.addSubcommand("write-secured-bulk", new WriteSecuredBulkCommand(clientFactory));
+        commandLine.addSubcommand("write-secured2-bulk", new WriteSecured2BulkCommand(clientFactory));
         commandLine.addSubcommand("write", new WriteCommand(clientFactory));
         commandLine.addSubcommand("stream-events", new StreamEventsCommand(clientFactory));
         commandLine.addSubcommand("smoke", new SmokeCommand(clientFactory));
@@ -518,6 +529,246 @@ public final class MxGatewayCli implements Callable<Integer> {
         }
     }
 
+    @Command(name = "read-bulk", description = "Invokes MXAccess ReadBulk (cached or snapshot per tag).")
+    static final class ReadBulkCommand extends GatewayCommand {
+        @Option(names = "--session-id", required = true, description = "Gateway session id.")
+        String sessionId;
+
+        @Option(names = "--server-handle", required = true, description = "MXAccess server handle.")
+        int serverHandle;
+
+        @Option(names = "--items", required = true, description = "Comma-separated tag addresses.")
+        String items;
+
+        @Option(names = "--timeout-ms", defaultValue = "0",
+                description = "Per-tag snapshot timeout in milliseconds (0 = worker default).")
+        int timeoutMs;
+
+        ReadBulkCommand(MxGatewayCliClientFactory clientFactory) {
+            super(clientFactory);
+        }
+
+        @Override
+        public Integer call() {
+            try (MxGatewayCliClient client = clientFactory.connect(common.resolved())) {
+                List<BulkReadResult> results =
+                        client.session(sessionId).readBulk(serverHandle, parseStringList(items), timeoutMs);
+                writeReadBulkOutput("read-bulk", common, json, results);
+            }
+            return 0;
+        }
+    }
+
+    @Command(name = "write-bulk", description = "Invokes MXAccess WriteBulk.")
+    static final class WriteBulkCommand extends GatewayCommand {
+        @Option(names = "--session-id", required = true, description = "Gateway session id.")
+        String sessionId;
+
+        @Option(names = "--server-handle", required = true, description = "MXAccess server handle.")
+        int serverHandle;
+
+        @Option(names = "--item-handles", required = true, description = "Comma-separated item handles.")
+        String itemHandles;
+
+        @Option(names = "--type", defaultValue = "string", description = "Value type for all entries.")
+        String type;
+
+        @Option(names = "--values", required = true, description = "Comma-separated values, one per item handle.")
+        String values;
+
+        @Option(names = "--user-id", defaultValue = "0", description = "MXAccess user id.")
+        int userId;
+
+        WriteBulkCommand(MxGatewayCliClientFactory clientFactory) {
+            super(clientFactory);
+        }
+
+        @Override
+        public Integer call() {
+            try (MxGatewayCliClient client = clientFactory.connect(common.resolved())) {
+                List<Integer> handles = parseIntList(itemHandles);
+                List<String> valueTexts = parseStringList(values);
+                if (handles.size() != valueTexts.size()) {
+                    throw new IllegalArgumentException(
+                            "item-handles count (" + handles.size() + ") does not match values count (" + valueTexts.size() + ")");
+                }
+                List<WriteBulkEntry> entries = new ArrayList<>(handles.size());
+                for (int i = 0; i < handles.size(); i++) {
+                    entries.add(WriteBulkEntry.newBuilder()
+                            .setItemHandle(handles.get(i))
+                            .setUserId(userId)
+                            .setValue(parseValue(type, valueTexts.get(i)))
+                            .build());
+                }
+                List<BulkWriteResult> results = client.session(sessionId).writeBulk(serverHandle, entries);
+                writeWriteBulkOutput("write-bulk", common, json, results);
+            }
+            return 0;
+        }
+    }
+
+    @Command(name = "write2-bulk", description = "Invokes MXAccess Write2Bulk (timestamped).")
+    static final class Write2BulkCommand extends GatewayCommand {
+        @Option(names = "--session-id", required = true, description = "Gateway session id.")
+        String sessionId;
+
+        @Option(names = "--server-handle", required = true, description = "MXAccess server handle.")
+        int serverHandle;
+
+        @Option(names = "--item-handles", required = true, description = "Comma-separated item handles.")
+        String itemHandles;
+
+        @Option(names = "--type", defaultValue = "string", description = "Value type for all entries.")
+        String type;
+
+        @Option(names = "--values", required = true, description = "Comma-separated values, one per item handle.")
+        String values;
+
+        @Option(names = "--timestamp", required = true, description = "ISO-8601 timestamp shared across all entries.")
+        String timestamp;
+
+        @Option(names = "--user-id", defaultValue = "0", description = "MXAccess user id.")
+        int userId;
+
+        Write2BulkCommand(MxGatewayCliClientFactory clientFactory) {
+            super(clientFactory);
+        }
+
+        @Override
+        public Integer call() {
+            try (MxGatewayCliClient client = clientFactory.connect(common.resolved())) {
+                List<Integer> handles = parseIntList(itemHandles);
+                List<String> valueTexts = parseStringList(values);
+                if (handles.size() != valueTexts.size()) {
+                    throw new IllegalArgumentException(
+                            "item-handles count (" + handles.size() + ") does not match values count (" + valueTexts.size() + ")");
+                }
+                MxValue timestampValue = MxValues.timestampValue(Instant.parse(timestamp));
+                List<Write2BulkEntry> entries = new ArrayList<>(handles.size());
+                for (int i = 0; i < handles.size(); i++) {
+                    entries.add(Write2BulkEntry.newBuilder()
+                            .setItemHandle(handles.get(i))
+                            .setUserId(userId)
+                            .setValue(parseValue(type, valueTexts.get(i)))
+                            .setTimestampValue(timestampValue)
+                            .build());
+                }
+                List<BulkWriteResult> results = client.session(sessionId).write2Bulk(serverHandle, entries);
+                writeWriteBulkOutput("write2-bulk", common, json, results);
+            }
+            return 0;
+        }
+    }
+
+    @Command(name = "write-secured-bulk", description = "Invokes MXAccess WriteSecuredBulk.")
+    static final class WriteSecuredBulkCommand extends GatewayCommand {
+        @Option(names = "--session-id", required = true, description = "Gateway session id.")
+        String sessionId;
+
+        @Option(names = "--server-handle", required = true, description = "MXAccess server handle.")
+        int serverHandle;
+
+        @Option(names = "--item-handles", required = true, description = "Comma-separated item handles.")
+        String itemHandles;
+
+        @Option(names = "--type", defaultValue = "string", description = "Value type for all entries.")
+        String type;
+
+        @Option(names = "--values", required = true, description = "Comma-separated values, one per item handle.")
+        String values;
+
+        @Option(names = "--current-user-id", defaultValue = "0", description = "MXAccess current user id.")
+        int currentUserId;
+
+        @Option(names = "--verifier-user-id", defaultValue = "0", description = "MXAccess verifier user id.")
+        int verifierUserId;
+
+        WriteSecuredBulkCommand(MxGatewayCliClientFactory clientFactory) {
+            super(clientFactory);
+        }
+
+        @Override
+        public Integer call() {
+            try (MxGatewayCliClient client = clientFactory.connect(common.resolved())) {
+                List<Integer> handles = parseIntList(itemHandles);
+                List<String> valueTexts = parseStringList(values);
+                if (handles.size() != valueTexts.size()) {
+                    throw new IllegalArgumentException(
+                            "item-handles count (" + handles.size() + ") does not match values count (" + valueTexts.size() + ")");
+                }
+                List<WriteSecuredBulkEntry> entries = new ArrayList<>(handles.size());
+                for (int i = 0; i < handles.size(); i++) {
+                    entries.add(WriteSecuredBulkEntry.newBuilder()
+                            .setItemHandle(handles.get(i))
+                            .setCurrentUserId(currentUserId)
+                            .setVerifierUserId(verifierUserId)
+                            .setValue(parseValue(type, valueTexts.get(i)))
+                            .build());
+                }
+                List<BulkWriteResult> results = client.session(sessionId).writeSecuredBulk(serverHandle, entries);
+                writeWriteBulkOutput("write-secured-bulk", common, json, results);
+            }
+            return 0;
+        }
+    }
+
+    @Command(name = "write-secured2-bulk", description = "Invokes MXAccess WriteSecured2Bulk.")
+    static final class WriteSecured2BulkCommand extends GatewayCommand {
+        @Option(names = "--session-id", required = true, description = "Gateway session id.")
+        String sessionId;
+
+        @Option(names = "--server-handle", required = true, description = "MXAccess server handle.")
+        int serverHandle;
+
+        @Option(names = "--item-handles", required = true, description = "Comma-separated item handles.")
+        String itemHandles;
+
+        @Option(names = "--type", defaultValue = "string", description = "Value type for all entries.")
+        String type;
+
+        @Option(names = "--values", required = true, description = "Comma-separated values, one per item handle.")
+        String values;
+
+        @Option(names = "--timestamp", required = true, description = "ISO-8601 timestamp shared across all entries.")
+        String timestamp;
+
+        @Option(names = "--current-user-id", defaultValue = "0", description = "MXAccess current user id.")
+        int currentUserId;
+
+        @Option(names = "--verifier-user-id", defaultValue = "0", description = "MXAccess verifier user id.")
+        int verifierUserId;
+
+        WriteSecured2BulkCommand(MxGatewayCliClientFactory clientFactory) {
+            super(clientFactory);
+        }
+
+        @Override
+        public Integer call() {
+            try (MxGatewayCliClient client = clientFactory.connect(common.resolved())) {
+                List<Integer> handles = parseIntList(itemHandles);
+                List<String> valueTexts = parseStringList(values);
+                if (handles.size() != valueTexts.size()) {
+                    throw new IllegalArgumentException(
+                            "item-handles count (" + handles.size() + ") does not match values count (" + valueTexts.size() + ")");
+                }
+                MxValue timestampValue = MxValues.timestampValue(Instant.parse(timestamp));
+                List<WriteSecured2BulkEntry> entries = new ArrayList<>(handles.size());
+                for (int i = 0; i < handles.size(); i++) {
+                    entries.add(WriteSecured2BulkEntry.newBuilder()
+                            .setItemHandle(handles.get(i))
+                            .setCurrentUserId(currentUserId)
+                            .setVerifierUserId(verifierUserId)
+                            .setValue(parseValue(type, valueTexts.get(i)))
+                            .setTimestampValue(timestampValue)
+                            .build());
+                }
+                List<BulkWriteResult> results = client.session(sessionId).writeSecured2Bulk(serverHandle, entries);
+                writeWriteBulkOutput("write-secured2-bulk", common, json, results);
+            }
+            return 0;
+        }
+    }
+
     @Command(name = "write", description = "Invokes MXAccess Write.")
     static final class WriteCommand extends GatewayCommand {
         @Option(names = "--session-id", required = true, description = "Gateway session id.")
@@ -760,6 +1011,16 @@ public final class MxGatewayCli implements Callable<Integer> {
 
         List<SubscribeResult> unsubscribeBulk(int serverHandle, List<Integer> itemHandles);
 
+        List<BulkReadResult> readBulk(int serverHandle, List<String> items, int timeoutMs);
+
+        List<BulkWriteResult> writeBulk(int serverHandle, List<WriteBulkEntry> entries);
+
+        List<BulkWriteResult> write2Bulk(int serverHandle, List<Write2BulkEntry> entries);
+
+        List<BulkWriteResult> writeSecuredBulk(int serverHandle, List<WriteSecuredBulkEntry> entries);
+
+        List<BulkWriteResult> writeSecured2Bulk(int serverHandle, List<WriteSecured2BulkEntry> entries);
+
         MxEventStream streamEventsAfter(long afterWorkerSequence);
     }
 
@@ -851,6 +1112,31 @@ public final class MxGatewayCli implements Callable<Integer> {
             return session.unsubscribeBulk(serverHandle, itemHandles);
         }
 
+        @Override
+        public List<BulkReadResult> readBulk(int serverHandle, List<String> items, int timeoutMs) {
+            return session.readBulk(serverHandle, items, timeoutMs);
+        }
+
+        @Override
+        public List<BulkWriteResult> writeBulk(int serverHandle, List<WriteBulkEntry> entries) {
+            return session.writeBulk(serverHandle, entries);
+        }
+
+        @Override
+        public List<BulkWriteResult> write2Bulk(int serverHandle, List<Write2BulkEntry> entries) {
+            return session.write2Bulk(serverHandle, entries);
+        }
+
+        @Override
+        public List<BulkWriteResult> writeSecuredBulk(int serverHandle, List<WriteSecuredBulkEntry> entries) {
+            return session.writeSecuredBulk(serverHandle, entries);
+        }
+
+        @Override
+        public List<BulkWriteResult> writeSecured2Bulk(int serverHandle, List<WriteSecured2BulkEntry> entries) {
+            return session.writeSecured2Bulk(serverHandle, entries);
+        }
+
         @Override
         public MxEventStream streamEventsAfter(long afterWorkerSequence) {
             return session.streamEventsAfter(afterWorkerSequence);
@@ -899,6 +1185,56 @@ public final class MxGatewayCli implements Callable<Integer> {
         return values;
     }
 
+    private static void writeWriteBulkOutput(
+            String command, CommonOptions common, boolean json, List<BulkWriteResult> results) {
+        PrintWriter out = common.spec.commandLine().getOut();
+        if (json) {
+            Map<String, Object> output = new LinkedHashMap<>();
+            output.put("command", command);
+            output.put("options", common.redactedJsonMap());
+            output.put("results", results.stream().map(MxGatewayCli::bulkWriteResultMap).toList());
+            out.println(jsonObject(output));
+            return;
+        }
+        out.println(results.size());
+    }
+
+    private static Map<String, Object> bulkWriteResultMap(BulkWriteResult result) {
+        Map<String, Object> values = new LinkedHashMap<>();
+        values.put("serverHandle", result.getServerHandle());
+        values.put("itemHandle", result.getItemHandle());
+        values.put("wasSuccessful", result.getWasSuccessful());
+        values.put("hresult", result.hasHresult() ? (Object) result.getHresult() : null);
+        values.put("errorMessage", result.getErrorMessage());
+        return values;
+    }
+
+    private static void writeReadBulkOutput(
+            String command, CommonOptions common, boolean json, List<BulkReadResult> results) {
+        PrintWriter out = common.spec.commandLine().getOut();
+        if (json) {
+            Map<String, Object> output = new LinkedHashMap<>();
+            output.put("command", command);
+            output.put("options", common.redactedJsonMap());
+            output.put("results", results.stream().map(MxGatewayCli::bulkReadResultMap).toList());
+            out.println(jsonObject(output));
+            return;
+        }
+        out.println(results.size());
+    }
+
+    private static Map<String, Object> bulkReadResultMap(BulkReadResult result) {
+        Map<String, Object> values = new LinkedHashMap<>();
+        values.put("serverHandle", result.getServerHandle());
+        values.put("tagAddress", result.getTagAddress());
+        values.put("itemHandle", result.getItemHandle());
+        values.put("wasSuccessful", result.getWasSuccessful());
+        values.put("wasCached", result.getWasCached());
+        values.put("quality", result.getQuality());
+        values.put("errorMessage", result.getErrorMessage());
+        return values;
+    }
+
     private static MxValue parseValue(String type, String text) {
         return switch (type) {
             case "bool" -> MxValues.boolValue(Boolean.parseBoolean(text));

clients/python/src/mxgateway_cli/commands.py

@@ -19,6 +19,7 @@ from mxgateway.errors import MxGatewayError
 from mxgateway.generated import mxaccess_gateway_pb2 as pb
 from mxgateway.options import ClientOptions
 from mxgateway.session import Session
+from mxgateway.values import to_mx_value
 from mxgateway.values import MxValueInput
 
 MAX_AGGREGATE_EVENTS = 10_000
@@ -186,6 +187,89 @@ def unsubscribe_bulk(**kwargs: Any) -> None:
     )
 
 
+@main.command("read-bulk")
+@gateway_options
+@click.option("--session-id", required=True, help="Gateway session id.")
+@click.option("--server-handle", required=True, type=int, help="MXAccess server handle.")
+@click.option("--items", required=True, help="Comma-separated MXAccess tag addresses.")
+@click.option("--timeout-ms", default=0, type=int, show_default=True,
+              help="Per-tag snapshot timeout in milliseconds. 0 = worker default.")
+@click.option("--correlation-id", default="", help="Client correlation id.")
+@click.option("--json", "output_json", is_flag=True, help="Emit JSON output.")
+def read_bulk(**kwargs: Any) -> None:
+    """Invoke MXAccess ReadBulk — cached value when advised, snapshot otherwise."""
+
+    _run(_read_bulk(**kwargs), output_json=kwargs["output_json"], secrets=_secrets(kwargs))
+
+
+@main.command("write-bulk")
+@gateway_options
+@click.option("--session-id", required=True, help="Gateway session id.")
+@click.option("--server-handle", required=True, type=int, help="MXAccess server handle.")
+@click.option("--item-handles", required=True, help="Comma-separated MXAccess item handles.")
+@click.option("--type", "value_type", default="string", show_default=True)
+@click.option("--values", required=True, help="Comma-separated values, one per item handle.")
+@click.option("--user-id", default=0, type=int, show_default=True)
+@click.option("--correlation-id", default="", help="Client correlation id.")
+@click.option("--json", "output_json", is_flag=True, help="Emit JSON output.")
+def write_bulk(**kwargs: Any) -> None:
+    """Invoke MXAccess WriteBulk — sequential Write per entry."""
+
+    _run(_write_bulk(**kwargs), output_json=kwargs["output_json"], secrets=_secrets(kwargs))
+
+
+@main.command("write2-bulk")
+@gateway_options
+@click.option("--session-id", required=True, help="Gateway session id.")
+@click.option("--server-handle", required=True, type=int, help="MXAccess server handle.")
+@click.option("--item-handles", required=True, help="Comma-separated MXAccess item handles.")
+@click.option("--type", "value_type", default="string", show_default=True)
+@click.option("--values", required=True, help="Comma-separated values, one per item handle.")
+@click.option("--timestamp", required=True, help="ISO-8601 timestamp shared across all entries.")
+@click.option("--user-id", default=0, type=int, show_default=True)
+@click.option("--correlation-id", default="", help="Client correlation id.")
+@click.option("--json", "output_json", is_flag=True, help="Emit JSON output.")
+def write2_bulk(**kwargs: Any) -> None:
+    """Invoke MXAccess Write2Bulk — timestamped sequential Write2 per entry."""
+
+    _run(_write2_bulk(**kwargs), output_json=kwargs["output_json"], secrets=_secrets(kwargs))
+
+
+@main.command("write-secured-bulk")
+@gateway_options
+@click.option("--session-id", required=True, help="Gateway session id.")
+@click.option("--server-handle", required=True, type=int, help="MXAccess server handle.")
+@click.option("--item-handles", required=True, help="Comma-separated MXAccess item handles.")
+@click.option("--type", "value_type", default="string", show_default=True)
+@click.option("--values", required=True, help="Comma-separated values, one per item handle.")
+@click.option("--current-user-id", default=0, type=int, show_default=True)
+@click.option("--verifier-user-id", default=0, type=int, show_default=True)
+@click.option("--correlation-id", default="", help="Client correlation id.")
+@click.option("--json", "output_json", is_flag=True, help="Emit JSON output.")
+def write_secured_bulk(**kwargs: Any) -> None:
+    """Invoke MXAccess WriteSecuredBulk — credential-sensitive."""
+
+    _run(_write_secured_bulk(**kwargs), output_json=kwargs["output_json"], secrets=_secrets(kwargs))
+
+
+@main.command("write-secured2-bulk")
+@gateway_options
+@click.option("--session-id", required=True, help="Gateway session id.")
+@click.option("--server-handle", required=True, type=int, help="MXAccess server handle.")
+@click.option("--item-handles", required=True, help="Comma-separated MXAccess item handles.")
+@click.option("--type", "value_type", default="string", show_default=True)
+@click.option("--values", required=True, help="Comma-separated values, one per item handle.")
+@click.option("--timestamp", required=True, help="ISO-8601 timestamp shared across all entries.")
+@click.option("--current-user-id", default=0, type=int, show_default=True)
+@click.option("--verifier-user-id", default=0, type=int, show_default=True)
+@click.option("--correlation-id", default="", help="Client correlation id.")
+@click.option("--json", "output_json", is_flag=True, help="Emit JSON output.")
+def write_secured2_bulk(**kwargs: Any) -> None:
+    """Invoke MXAccess WriteSecured2Bulk — timestamped + credential-sensitive."""
+
+    _run(_write_secured2_bulk(**kwargs), output_json=kwargs["output_json"], secrets=_secrets(kwargs))
+
+
 @main.command("stream-events")
 @gateway_options
 @click.option("--session-id", required=True, help="Gateway session id.")
@@ -340,6 +424,120 @@ async def _unsubscribe_bulk(**kwargs: Any) -> dict[str, Any]:
         return {"results": [_message_dict(result) for result in results]}
 
 
+async def _read_bulk(**kwargs: Any) -> dict[str, Any]:
+    async with await _connect(kwargs) as client:
+        session = _session(client, kwargs["session_id"])
+        results = await session.read_bulk(
+            kwargs["server_handle"],
+            _parse_string_list(kwargs["items"]),
+            timeout_ms=kwargs["timeout_ms"],
+            correlation_id=kwargs["correlation_id"],
+        )
+        return {"results": [_message_dict(result) for result in results]}
+
+
+def _build_write_bulk_entries(kwargs: dict[str, Any]):
+    """Build (item_handle, MxValue) pairs for the bulk-write families.
+
+    The CLI accepts a single ``--type`` plus ``--values`` (comma-separated
+    string-encoded values, one per ``--item-handles`` entry). Returns the
+    parsed item-handle list and the per-entry MxValue protobuf instances —
+    callers wrap these into the appropriate per-entry message type.
+    """
+
+    handles = _parse_int_list(kwargs["item_handles"])
+    value_texts = _parse_string_list(kwargs["values"])
+    if len(handles) != len(value_texts):
+        raise click.UsageError(
+            f"item-handles count ({len(handles)}) does not match values count ({len(value_texts)})",
+        )
+    parsed = [_parse_value(text, kwargs["value_type"]) for text in value_texts]
+    values = [to_mx_value(v) for v in parsed]
+    return handles, values
+
+
+async def _write_bulk(**kwargs: Any) -> dict[str, Any]:
+    handles, values = _build_write_bulk_entries(kwargs)
+    entries = [
+        pb.WriteBulkEntry(item_handle=handle, user_id=kwargs["user_id"], value=value)
+        for handle, value in zip(handles, values)
+    ]
+    async with await _connect(kwargs) as client:
+        session = _session(client, kwargs["session_id"])
+        results = await session.write_bulk(
+            kwargs["server_handle"],
+            entries,
+            correlation_id=kwargs["correlation_id"],
+        )
+        return {"results": [_message_dict(result) for result in results]}
+
+
+async def _write2_bulk(**kwargs: Any) -> dict[str, Any]:
+    handles, values = _build_write_bulk_entries(kwargs)
+    timestamp_value = to_mx_value(_parse_datetime(kwargs["timestamp"]))
+    entries = [
+        pb.Write2BulkEntry(
+            item_handle=handle,
+            user_id=kwargs["user_id"],
+            value=value,
+            timestamp_value=timestamp_value,
+        )
+        for handle, value in zip(handles, values)
+    ]
+    async with await _connect(kwargs) as client:
+        session = _session(client, kwargs["session_id"])
+        results = await session.write2_bulk(
+            kwargs["server_handle"],
+            entries,
+            correlation_id=kwargs["correlation_id"],
+        )
+        return {"results": [_message_dict(result) for result in results]}
+
+
+async def _write_secured_bulk(**kwargs: Any) -> dict[str, Any]:
+    handles, values = _build_write_bulk_entries(kwargs)
+    entries = [
+        pb.WriteSecuredBulkEntry(
+            item_handle=handle,
+            current_user_id=kwargs["current_user_id"],
+            verifier_user_id=kwargs["verifier_user_id"],
+            value=value,
+        )
+        for handle, value in zip(handles, values)
+    ]
+    async with await _connect(kwargs) as client:
+        session = _session(client, kwargs["session_id"])
+        results = await session.write_secured_bulk(
+            kwargs["server_handle"],
+            entries,
+            correlation_id=kwargs["correlation_id"],
+        )
+        return {"results": [_message_dict(result) for result in results]}
+
+
+async def _write_secured2_bulk(**kwargs: Any) -> dict[str, Any]:
+    handles, values = _build_write_bulk_entries(kwargs)
+    timestamp_value = to_mx_value(_parse_datetime(kwargs["timestamp"]))
+    entries = [
+        pb.WriteSecured2BulkEntry(
+            item_handle=handle,
+            current_user_id=kwargs["current_user_id"],
+            verifier_user_id=kwargs["verifier_user_id"],
+            value=value,
+            timestamp_value=timestamp_value,
+        )
+        for handle, value in zip(handles, values)
+    ]
+    async with await _connect(kwargs) as client:
+        session = _session(client, kwargs["session_id"])
+        results = await session.write_secured2_bulk(
+            kwargs["server_handle"],
+            entries,
+            correlation_id=kwargs["correlation_id"],
+        )
+        return {"results": [_message_dict(result) for result in results]}
+
+
 async def _stream_events(**kwargs: Any) -> dict[str, Any]:
     async with await _connect(kwargs) as client:
         session = _session(client, kwargs["session_id"])

clients/rust/crates/mxgw-cli/src/main.rs

@@ -18,7 +18,8 @@ use futures_util::StreamExt;
 use mxgateway_client::generated::galaxy_repository::v1::DeployEvent;
 use mxgateway_client::generated::mxaccess_gateway::v1::{
     CloseSessionRequest, MxCommand, MxCommandKind, MxCommandRequest, MxEvent, MxEventFamily,
-    MxValue as ProtoMxValue, OpenSessionRequest, PingCommand, StreamEventsRequest,
+    MxValue as ProtoMxValue, OpenSessionRequest, PingCommand, StreamEventsRequest, Write2BulkEntry,
+    WriteBulkEntry, WriteSecured2BulkEntry, WriteSecuredBulkEntry,
 };
 use mxgateway_client::{
     ApiKey, ClientOptions, Error, GalaxyClient, GatewayClient, MxValue, MxValueProjection,
@@ -127,6 +128,109 @@ enum Command {
         #[arg(long)]
         json: bool,
     },
+    /// Snapshot the current value for each requested tag. Cached
+    /// OnDataChange values are returned for tags that are already advised
+    /// without touching the existing subscription; otherwise the worker
+    /// takes a one-shot AddItem + Advise + UnAdvise + RemoveItem lifecycle.
+    ReadBulk {
+        #[command(flatten)]
+        connection: ConnectionArgs,
+        #[arg(long)]
+        session_id: String,
+        #[arg(long)]
+        server_handle: i32,
+        #[arg(long, value_delimiter = ',')]
+        items: Vec<String>,
+        /// Per-tag snapshot timeout in milliseconds. `0` uses the worker default (1000 ms).
+        #[arg(long, default_value_t = 0)]
+        timeout_ms: u32,
+        #[arg(long)]
+        json: bool,
+    },
+    /// Bulk Write — one MXAccess Write per (item_handle, value) pair.
+    WriteBulk {
+        #[command(flatten)]
+        connection: ConnectionArgs,
+        #[arg(long)]
+        session_id: String,
+        #[arg(long)]
+        server_handle: i32,
+        #[arg(long, value_delimiter = ',')]
+        item_handles: Vec<i32>,
+        #[arg(long, value_enum)]
+        value_type: CliValueType,
+        #[arg(long, value_delimiter = ',')]
+        values: Vec<String>,
+        #[arg(long, default_value_t = 0)]
+        user_id: i32,
+        #[arg(long)]
+        json: bool,
+    },
+    /// Bulk Write2 — timestamped variant; the timestamp applies to all entries.
+    Write2Bulk {
+        #[command(flatten)]
+        connection: ConnectionArgs,
+        #[arg(long)]
+        session_id: String,
+        #[arg(long)]
+        server_handle: i32,
+        #[arg(long, value_delimiter = ',')]
+        item_handles: Vec<i32>,
+        #[arg(long, value_enum)]
+        value_type: CliValueType,
+        #[arg(long, value_delimiter = ',')]
+        values: Vec<String>,
+        #[arg(long)]
+        timestamp: String,
+        #[arg(long, default_value_t = 0)]
+        user_id: i32,
+        #[arg(long)]
+        json: bool,
+    },
+    /// Bulk WriteSecured.
+    WriteSecuredBulk {
+        #[command(flatten)]
+        connection: ConnectionArgs,
+        #[arg(long)]
+        session_id: String,
+        #[arg(long)]
+        server_handle: i32,
+        #[arg(long, value_delimiter = ',')]
+        item_handles: Vec<i32>,
+        #[arg(long, value_enum)]
+        value_type: CliValueType,
+        #[arg(long, value_delimiter = ',')]
+        values: Vec<String>,
+        #[arg(long, default_value_t = 0)]
+        current_user_id: i32,
+        #[arg(long, default_value_t = 0)]
+        verifier_user_id: i32,
+        #[arg(long)]
+        json: bool,
+    },
+    /// Bulk WriteSecured2 — timestamped + verified.
+    WriteSecured2Bulk {
+        #[command(flatten)]
+        connection: ConnectionArgs,
+        #[arg(long)]
+        session_id: String,
+        #[arg(long)]
+        server_handle: i32,
+        #[arg(long, value_delimiter = ',')]
+        item_handles: Vec<i32>,
+        #[arg(long, value_enum)]
+        value_type: CliValueType,
+        #[arg(long, value_delimiter = ',')]
+        values: Vec<String>,
+        #[arg(long)]
+        timestamp: String,
+        #[arg(long, default_value_t = 0)]
+        current_user_id: i32,
+        #[arg(long, default_value_t = 0)]
+        verifier_user_id: i32,
+        #[arg(long)]
+        json: bool,
+    },
     StreamEvents {
         #[command(flatten)]
         connection: ConnectionArgs,
@@ -429,6 +533,136 @@ async fn run(cli: Cli) -> Result<(), Error> {
                 .await?;
             print_bulk_results("unsubscribe-bulk", &results, json);
         }
+        Command::ReadBulk {
+            connection,
+            session_id,
+            server_handle,
+            items,
+            timeout_ms,
+            json,
+        } => {
+            let session = session_for(connection, session_id).await?;
+            let results = session.read_bulk(server_handle, items, timeout_ms).await?;
+            print_read_bulk_results("read-bulk", &results, json);
+        }
+        Command::WriteBulk {
+            connection,
+            session_id,
+            server_handle,
+            item_handles,
+            value_type,
+            values,
+            user_id,
+            json,
+        } => {
+            let entries = build_write_bulk_entries(&item_handles, value_type, &values)?;
+            let session = session_for(connection, session_id).await?;
+            let results = session
+                .write_bulk(
+                    server_handle,
+                    entries
+                        .into_iter()
+                        .map(|(item_handle, value)| WriteBulkEntry {
+                            item_handle,
+                            value: Some(value),
+                            user_id,
+                        })
+                        .collect(),
+                )
+                .await?;
+            print_write_bulk_results("write-bulk", &results, json);
+        }
+        Command::Write2Bulk {
+            connection,
+            session_id,
+            server_handle,
+            item_handles,
+            value_type,
+            values,
+            timestamp,
+            user_id,
+            json,
+        } => {
+            let entries = build_write_bulk_entries(&item_handles, value_type, &values)?;
+            let timestamp_value: ProtoMxValue = MxValue::string(timestamp).into_proto();
+            let session = session_for(connection, session_id).await?;
+            let results = session
+                .write2_bulk(
+                    server_handle,
+                    entries
+                        .into_iter()
+                        .map(|(item_handle, value)| Write2BulkEntry {
+                            item_handle,
+                            value: Some(value),
+                            timestamp_value: Some(timestamp_value.clone()),
+                            user_id,
+                        })
+                        .collect(),
+                )
+                .await?;
+            print_write_bulk_results("write2-bulk", &results, json);
+        }
+        Command::WriteSecuredBulk {
+            connection,
+            session_id,
+            server_handle,
+            item_handles,
+            value_type,
+            values,
+            current_user_id,
+            verifier_user_id,
+            json,
+        } => {
+            let entries = build_write_bulk_entries(&item_handles, value_type, &values)?;
+            let session = session_for(connection, session_id).await?;
+            let results = session
+                .write_secured_bulk(
+                    server_handle,
+                    entries
+                        .into_iter()
+                        .map(|(item_handle, value)| WriteSecuredBulkEntry {
+                            item_handle,
+                            value: Some(value),
+                            current_user_id,
+                            verifier_user_id,
+                        })
+                        .collect(),
+                )
+                .await?;
+            print_write_bulk_results("write-secured-bulk", &results, json);
+        }
+        Command::WriteSecured2Bulk {
+            connection,
+            session_id,
+            server_handle,
+            item_handles,
+            value_type,
+            values,
+            timestamp,
+            current_user_id,
+            verifier_user_id,
+            json,
+        } => {
+            let entries = build_write_bulk_entries(&item_handles, value_type, &values)?;
+            let timestamp_value: ProtoMxValue = MxValue::string(timestamp).into_proto();
+            let session = session_for(connection, session_id).await?;
+            let results = session
+                .write_secured2_bulk(
+                    server_handle,
+                    entries
+                        .into_iter()
+                        .map(|(item_handle, value)| WriteSecured2BulkEntry {
+                            item_handle,
+                            value: Some(value),
+                            timestamp_value: Some(timestamp_value.clone()),
+                            current_user_id,
+                            verifier_user_id,
+                        })
+                        .collect(),
+                )
+                .await?;
+            print_write_bulk_results("write-secured2-bulk", &results, json);
+        }
         Command::StreamEvents {
             connection,
             session_id,
@@ -784,6 +1018,89 @@ fn print_bulk_results(
     }
 }
 
+fn print_write_bulk_results(
+    operation: &str,
+    results: &[mxgateway_client::generated::mxaccess_gateway::v1::BulkWriteResult],
+    use_json: bool,
+) {
+    if use_json {
+        let results_json: Vec<_> = results
+            .iter()
+            .map(|result| {
+                json!({
+                    "serverHandle": result.server_handle,
+                    "itemHandle": result.item_handle,
+                    "wasSuccessful": result.was_successful,
+                    "hresult": result.hresult,
+                    "errorMessage": result.error_message,
+                })
+            })
+            .collect();
+        println!(
+            "{}",
+            json!({ "operation": operation, "results": results_json })
+        );
+    } else {
+        println!("{}", results.len());
+    }
+}
+
+fn print_read_bulk_results(
+    operation: &str,
+    results: &[mxgateway_client::generated::mxaccess_gateway::v1::BulkReadResult],
+    use_json: bool,
+) {
+    if use_json {
+        let results_json: Vec<_> = results
+            .iter()
+            .map(|result| {
+                json!({
+                    "serverHandle": result.server_handle,
+                    "tagAddress": result.tag_address,
+                    "itemHandle": result.item_handle,
+                    "wasSuccessful": result.was_successful,
+                    "wasCached": result.was_cached,
+                    "quality": result.quality,
+                    "errorMessage": result.error_message,
+                })
+            })
+            .collect();
+        println!(
+            "{}",
+            json!({ "operation": operation, "results": results_json })
+        );
+    } else {
+        println!("{}", results.len());
+    }
+}
+
+/// Pairs each parsed item handle with its parsed MxValue (proto form) so a
+/// single helper can build the four bulk-write families without each branch
+/// repeating the length check and per-value parsing.
+fn build_write_bulk_entries(
+    item_handles: &[i32],
+    value_type: CliValueType,
+    values: &[String],
+) -> Result<Vec<(i32, mxgateway_client::generated::mxaccess_gateway::v1::MxValue)>, Error> {
+    if item_handles.len() != values.len() {
+        return Err(Error::InvalidArgument {
+            name: "values".to_owned(),
+            detail: format!(
+                "item-handles count ({}) does not match values count ({})",
+                item_handles.len(),
+                values.len()
+            ),
+        });
+    }
+    item_handles
+        .iter()
+        .zip(values.iter())
+        .map(|(handle, value)| {
+            parse_value(value_type, value).map(|wrapper| (*handle, wrapper.into_proto()))
+        })
+        .collect()
+}
+
 fn parse_value(value_type: CliValueType, value: &str) -> Result<MxValue, Error> {
     let parsed = match value_type {
         CliValueType::Bool => MxValue::bool(parse_cli_value(value)?),

scripts/run-client-e2e-tests.ps1

@@ -33,6 +33,12 @@ param(
     [int]$BulkTagCount = 6,
     [switch]$SkipStream,
     [switch]$SkipBulk,
+    # Skip the bulk read+write coverage that runs alongside the existing
+    # subscribe-bulk phase. The read-bulk phase confirms cached-path
+    # semantics against tags left advised by subscribe-bulk (was_cached
+    # = true); the write-bulk phase runs when -VerifyWrite is set and
+    # exercises the BulkWriteResult shape against the writable tag.
+    [switch]$SkipReadWriteBulk,
     # Write round-trip. Opt-in because it mutates live tag state: it writes a
     # sentinel value to -WriteAttribute and asserts an OnWriteComplete event
     # confirms the write reached the MXAccess provider.
@@ -400,7 +406,18 @@ function Get-BulkResults {
         return @(Get-PropertyValue -Object $Json -Names @("results"))
     }
 
-    $replyName = if ($Operation -eq "subscribe-bulk") { "subscribeBulk" } else { "unsubscribeBulk" }
+    # .NET emits the full MxCommandReply via protobuf JSON, with results
+    # nested under a per-command field name.
+    $replyName = switch ($Operation) {
+        "subscribe-bulk" { "subscribeBulk" }
+        "unsubscribe-bulk" { "unsubscribeBulk" }
+        "read-bulk" { "readBulk" }
+        "write-bulk" { "writeBulk" }
+        "write2-bulk" { "write2Bulk" }
+        "write-secured-bulk" { "writeSecuredBulk" }
+        "write-secured2-bulk" { "writeSecured2Bulk" }
+        default { $Operation }
+    }
     $reply = Get-PropertyValue -Object $Json -Names @($replyName)
     return @(Get-PropertyValue -Object $reply -Names @("results"))
 }
@@ -478,6 +495,13 @@ function Get-ClientCommand {
                 $arguments += @("--session-id", $Values.sessionId, "--server-handle", "$($Values.serverHandle)", "--items", $Values.items)
             } elseif ($Operation -eq "unsubscribe-bulk") {
                 $arguments += @("--session-id", $Values.sessionId, "--server-handle", "$($Values.serverHandle)", "--item-handles", $Values.itemHandles)
+            } elseif ($Operation -eq "read-bulk") {
+                $arguments += @("--session-id", $Values.sessionId, "--server-handle", "$($Values.serverHandle)", "--items", $Values.items)
+                if ($Values.ContainsKey("timeoutMs")) { $arguments += @("--timeout-ms", "$($Values.timeoutMs)") }
+            } elseif ($Operation -eq "write-bulk") {
+                $arguments += @("--session-id", $Values.sessionId, "--server-handle", "$($Values.serverHandle)",
+                    "--item-handles", $Values.itemHandles, "--type", $Values.valueType, "--values", $Values.values)
+                if ($Values.ContainsKey("userId")) { $arguments += @("--user-id", "$($Values.userId)") }
             } elseif ($Operation -eq "write") {
                 $arguments += @("--session-id", $Values.sessionId, "--server-handle", "$($Values.serverHandle)", "--item-handle", "$($Values.itemHandle)", "--type", $Values.valueType, "--value", $Values.value)
             } elseif ($Operation -eq "stream-events") {
@@ -507,6 +531,13 @@ function Get-ClientCommand {
                 $arguments += @("-session-id", $Values.sessionId, "-server-handle", "$($Values.serverHandle)", "-items", $Values.items)
             } elseif ($Operation -eq "unsubscribe-bulk") {
                 $arguments += @("-session-id", $Values.sessionId, "-server-handle", "$($Values.serverHandle)", "-item-handles", $Values.itemHandles)
+            } elseif ($Operation -eq "read-bulk") {
+                $arguments += @("-session-id", $Values.sessionId, "-server-handle", "$($Values.serverHandle)", "-items", $Values.items)
+                if ($Values.ContainsKey("timeoutMs")) { $arguments += @("-timeout-ms", "$($Values.timeoutMs)") }
+            } elseif ($Operation -eq "write-bulk") {
+                $arguments += @("-session-id", $Values.sessionId, "-server-handle", "$($Values.serverHandle)",
+                    "-item-handles", $Values.itemHandles, "-type", $Values.valueType, "-values", $Values.values)
+                if ($Values.ContainsKey("userId")) { $arguments += @("-user-id", "$($Values.userId)") }
             } elseif ($Operation -eq "write") {
                 $arguments += @("-session-id", $Values.sessionId, "-server-handle", "$($Values.serverHandle)", "-item-handle", "$($Values.itemHandle)", "-type", $Values.valueType, "-value", $Values.value)
             } elseif ($Operation -eq "stream-events") {
@@ -535,6 +566,14 @@ function Get-ClientCommand {
                 $arguments += @("--session-id", $Values.sessionId, "--server-handle", "$($Values.serverHandle)", "--items", $Values.items)
             } elseif ($Operation -eq "unsubscribe-bulk") {
                 $arguments += @("--session-id", $Values.sessionId, "--server-handle", "$($Values.serverHandle)", "--item-handles", $Values.itemHandles)
+            } elseif ($Operation -eq "read-bulk") {
+                $arguments += @("--session-id", $Values.sessionId, "--server-handle", "$($Values.serverHandle)", "--items", $Values.items)
+                if ($Values.ContainsKey("timeoutMs")) { $arguments += @("--timeout-ms", "$($Values.timeoutMs)") }
+            } elseif ($Operation -eq "write-bulk") {
+                # Rust uses --value-type for the type flag.
+                $arguments += @("--session-id", $Values.sessionId, "--server-handle", "$($Values.serverHandle)",
+                    "--item-handles", $Values.itemHandles, "--value-type", $Values.valueType, "--values", $Values.values)
+                if ($Values.ContainsKey("userId")) { $arguments += @("--user-id", "$($Values.userId)") }
             } elseif ($Operation -eq "write") {
                 # Rust names the type flag --value-type, unlike the other CLIs.
                 $arguments += @("--session-id", $Values.sessionId, "--server-handle", "$($Values.serverHandle)", "--item-handle", "$($Values.itemHandle)", "--value-type", $Values.valueType, "--value", $Values.value)
@@ -565,6 +604,13 @@ function Get-ClientCommand {
                 $arguments += @("--session-id", $Values.sessionId, "--server-handle", "$($Values.serverHandle)", "--items", $Values.items)
             } elseif ($Operation -eq "unsubscribe-bulk") {
                 $arguments += @("--session-id", $Values.sessionId, "--server-handle", "$($Values.serverHandle)", "--item-handles", $Values.itemHandles)
+            } elseif ($Operation -eq "read-bulk") {
+                $arguments += @("--session-id", $Values.sessionId, "--server-handle", "$($Values.serverHandle)", "--items", $Values.items)
+                if ($Values.ContainsKey("timeoutMs")) { $arguments += @("--timeout-ms", "$($Values.timeoutMs)") }
+            } elseif ($Operation -eq "write-bulk") {
+                $arguments += @("--session-id", $Values.sessionId, "--server-handle", "$($Values.serverHandle)",
+                    "--item-handles", $Values.itemHandles, "--type", $Values.valueType, "--values", $Values.values)
+                if ($Values.ContainsKey("userId")) { $arguments += @("--user-id", "$($Values.userId)") }
             } elseif ($Operation -eq "write") {
                 $arguments += @("--session-id", $Values.sessionId, "--server-handle", "$($Values.serverHandle)", "--item-handle", "$($Values.itemHandle)", "--type", $Values.valueType, "--value", $Values.value)
             } elseif ($Operation -eq "stream-events") {
@@ -597,6 +643,13 @@ function Get-ClientCommand {
                 $cliArgs += @("--session-id", $Values.sessionId, "--server-handle", "$($Values.serverHandle)", "--items", $Values.items)
             } elseif ($Operation -eq "unsubscribe-bulk") {
                 $cliArgs += @("--session-id", $Values.sessionId, "--server-handle", "$($Values.serverHandle)", "--item-handles", $Values.itemHandles)
+            } elseif ($Operation -eq "read-bulk") {
+                $cliArgs += @("--session-id", $Values.sessionId, "--server-handle", "$($Values.serverHandle)", "--items", $Values.items)
+                if ($Values.ContainsKey("timeoutMs")) { $cliArgs += @("--timeout-ms", "$($Values.timeoutMs)") }
+            } elseif ($Operation -eq "write-bulk") {
+                $cliArgs += @("--session-id", $Values.sessionId, "--server-handle", "$($Values.serverHandle)",
+                    "--item-handles", $Values.itemHandles, "--type", $Values.valueType, "--values", $Values.values)
+                if ($Values.ContainsKey("userId")) { $cliArgs += @("--user-id", "$($Values.userId)") }
             } elseif ($Operation -eq "write") {
                 $cliArgs += @("--session-id", $Values.sessionId, "--server-handle", "$($Values.serverHandle)", "--item-handle", "$($Values.itemHandle)", "--type", $Values.valueType, "--value", $Values.value)
             } elseif ($Operation -eq "stream-events") {
@@ -649,6 +702,23 @@ function Get-DryRunReply {
             })
             return [pscustomobject]@{ unsubscribeBulk = [pscustomobject]@{ results = $results }; results = $results }
         }
+        "read-bulk" {
+            $results = @($Values.items -split "," | ForEach-Object -Begin { $index = 1 } -Process {
+                [pscustomobject]@{
+                    itemHandle = $index++
+                    tagAddress = $_
+                    wasSuccessful = $true
+                    wasCached = $true
+                }
+            })
+            return [pscustomobject]@{ readBulk = [pscustomobject]@{ results = $results }; results = $results }
+        }
+        "write-bulk" {
+            $results = @($Values.itemHandles -split "," | ForEach-Object {
+                [pscustomobject]@{ itemHandle = [int]$_; wasSuccessful = $true }
+            })
+            return [pscustomobject]@{ writeBulk = [pscustomobject]@{ results = $results }; results = $results }
+        }
         "stream-events" {
             # Synthesize an OnDataChange (carrying the written value) and an
             # OnWriteComplete so the write round-trip assertion passes under
@@ -839,6 +909,28 @@ function Invoke-ClientFlow {
                     writeCompleteObserved = $true
                     echoObserved = ($null -ne $echoEvent)
                 }
+
+                # WriteBulk smoke: single-entry batch against the same writable
+                # tag. Exercises the BulkWriteResult wire format end-to-end
+                # without complicating the OnWriteComplete echo assertion that
+                # the single-item write phase already verified above. Pinned
+                # to a different sentinel value so a subsequent read-bulk
+                # against the same tag would see the bulk write's effect.
+                if (-not $SkipReadWriteBulk) {
+                    $bulkSentinel = $sentinelValue + 1
+                    $writeBulkJson = Invoke-ClientOperation -Client $Client -Operation "write-bulk" -Values @{
+                        sessionId = $sessionId
+                        serverHandle = $serverHandle
+                        itemHandles = "$writeItemHandle"
+                        valueType = $WriteType
+                        values = "$bulkSentinel"
+                        userId = 0
+                    }
+                    $writeBulkResults = @(Get-BulkResults -Client $Client -Operation "write-bulk" -Json $writeBulkJson)
+                    Assert-BulkResults -Client $Client -Operation "write-bulk" -Results $writeBulkResults -ExpectedCount 1
+                    $clientResult.write.writeBulkValue = $bulkSentinel
+                    $clientResult.write.writeBulkResultCount = $writeBulkResults.Count
+                }
             }
         }
 
@@ -857,6 +949,40 @@ function Invoke-ClientFlow {
                 throw "$Client subscribe-bulk returned $($bulkItemHandles.Count) usable item handle(s); expected $($bulkTags.Count)."
             }
 
+            # ReadBulk over the already-advised tags: every result must come
+            # from the per-session value cache (was_cached = true). Confirms
+            # the gateway/worker/cache wiring serves cached values for tags
+            # the caller did not create the subscription for.
+            $readBulkSummary = $null
+            if (-not $SkipReadWriteBulk) {
+                $readBulkJson = Invoke-ClientOperation -Client $Client -Operation "read-bulk" -Values @{
+                    sessionId = $sessionId
+                    serverHandle = $serverHandle
+                    items = $bulkItems
+                    timeoutMs = 1500
+                }
+                $readResults = @(Get-BulkResults -Client $Client -Operation "read-bulk" -Json $readBulkJson)
+                Assert-BulkResults -Client $Client -Operation "read-bulk" -Results $readResults -ExpectedCount $bulkTags.Count
+                $cachedCount = @($readResults | Where-Object {
+                    [bool](Get-PropertyValue -Object $_ -Names @("wasCached", "was_cached"))
+                }).Count
+                # Allow up to one snapshot fallback per batch: a freshly
+                # advised tag may not have an OnDataChange cached yet if it
+                # hasn't pushed an update in the small window between
+                # subscribe-bulk and read-bulk. Anything beyond that means
+                # the cached-path optimization is broken.
+                $maxSnapshotFallbacks = 1
+                if ($cachedCount -lt ($readResults.Count - $maxSnapshotFallbacks)) {
+                    throw ("$Client read-bulk only returned $cachedCount cached result(s) " +
+                        "out of $($readResults.Count); the cache-then-snapshot fork must " +
+                        "serve cached values for already-advised tags.")
+                }
+                $readBulkSummary = [ordered]@{
+                    tagCount = $readResults.Count
+                    cachedCount = $cachedCount
+                }
+            }
+
             $unsubscribeBulkJson = Invoke-ClientOperation -Client $Client -Operation "unsubscribe-bulk" -Values @{
                 sessionId = $sessionId
                 serverHandle = $serverHandle
@@ -870,6 +996,7 @@ function Invoke-ClientFlow {
                 subscribedCount = $subscribeResults.Count
                 unsubscribedCount = $unsubscribeResults.Count
                 itemHandles = $bulkItemHandles
+                readBulk = $readBulkSummary
             }
         }