fix(send): preserve client leadership

Commit `8ba312d`

凡辞 · 2026-07-22 12:02:23 -0400 EDT

parent 935c484

fix(send): preserve client leadership

3 files changed, +34, -4

M CHANGELOG.md

M src/ipc.zig

M src/main.zig

M CHANGELOG.md

+1, -0

......@@ -19,6 +19,7 @@ Use spec: https://common-changelog.org/
- `zmx run` will now detect heredocs and add the completion marker to a newline
- Race between `zmx kill X; zmx run X`
- Improved claim leader detection
- `zmx send` no longer claims client leadership or triggers a resize probe
- Improved `ctrl+\` key detection
- Support for linux kernel < 4.11 by avoiding statx calls

M src/ipc.zig

+5, -3

......@@ -22,15 +22,16 @@ pub const Tag = enum(u8) {
    LabelSet = 15,
    LabelClear = 16,
    LabelData = 17,
    Send = 18,
    // Non-exhaustive: this enum comes off the wire via bytesToValue and
    // @enumFromInt, so out-of-range values (14-255) are representable
    // @enumFromInt, so out-of-range values are representable
    // rather than UB. Switches must handle `_` (unknown tag).
    _,
};

comptime {
    if (@typeInfo(Tag).@"enum".is_exhaustive) @compileError(
        "ipc.Tag must stay non-exhaustive — old daemons rely on `_` to ignore unknown tags",
        "ipc.Tag must stay non-exhaustive -- old daemons rely on `_` to ignore unknown tags",
    );
}

......@@ -57,7 +58,7 @@ pub fn getTerminalSize(fd: i32) Resize {
pub const MAX_CMD_LEN = 256;
pub const MAX_CWD_LEN = 256;

/// Frozen wire shape. Do NOT add fields — new stats go in new `Tag` values
/// Frozen wire shape. Do NOT add fields! New stats go in new `Tag` values
/// so old daemons (whose `_` arm ignores unknown tags) stay reachable.
/// Changing `@sizeOf(Info)` breaks `zmx list` against running daemons.
pub const Info = extern struct {

......@@ -306,6 +307,7 @@ test "Tag wire values are frozen" {
        .{ Tag.Run, 9 },       .{ Tag.Ack, 10 },          .{ Tag.Switch, 11 },
        .{ Tag.Write, 12 },    .{ Tag.TaskComplete, 13 }, .{ Tag.LabelGet, 14 },
        .{ Tag.LabelSet, 15 }, .{ Tag.LabelClear, 16 },   .{ Tag.LabelData, 17 },
        .{ Tag.Send, 18 },
    }) |p| try std.testing.expectEqual(@as(u8, p[1]), @intFromEnum(p[0]));
}

M src/main.zig

+28, -1

......@@ -287,7 +287,7 @@ pub fn main() !void {
            error.NameTooLong => return socket.printSessionNameTooLong(sesh, cfg.socket_dir),
            error.OutOfMemory => return err,
        };
        return send(&cfg, sesh, socket_path, text_parts.items, .Input);
        return send(&cfg, sesh, socket_path, text_parts.items, .Send);
    } else if (std.mem.eql(u8, cmd, "print") or std.mem.eql(u8, cmd, "p")) {
        const session_name = args.next() orelse "";
        if (std.mem.eql(u8, session_name, "--help") or std.mem.eql(u8, session_name, "-h")) {

......@@ -1046,6 +1046,11 @@ const Daemon = struct {
        }
    }

    /// Queue input from `zmx send` without changing interactive client leadership.
    pub fn handleSend(self: *Daemon, payload: []const u8) void {
        self.queuePtyInput(payload);
    }

    pub fn handleSwitch(self: *Daemon, session_name: []const u8) !void {
        for (self.clients.items) |client| {
            if (self.leader_client_fd == client.socket_fd) {

......@@ -1371,6 +1376,27 @@ const Daemon = struct {
    }
};

test "send queues PTY input without changing leader" {
    const alloc = std.testing.allocator;
    var daemon = Daemon{
        .cfg = undefined,
        .alloc = alloc,
        .clients = .empty,
        .leader_client_fd = 42,
        .session_name = "test",
        .socket_path = "",
        .running = true,
        .pid = 0,
        .created_at = 0,
    };
    defer daemon.pty_write_buf.deinit(alloc);

    daemon.handleSend("hello");

    try std.testing.expectEqual(@as(?i32, 42), daemon.leader_client_fd);
    try std.testing.expectEqualStrings("hello", daemon.pty_write_buf.items);
}

fn printVersion(cfg: *Cfg) !void {
    var buf: [256]u8 = undefined;
    var w = std.fs.File.stdout().writer(&buf);

......@@ -3042,6 +3068,7 @@ fn daemonLoop(daemon: *Daemon, server_sock_fd: i32, pty_fd: i32) !void {
                while (client.read_buf.next()) |msg| {
                    switch (msg.header.tag) {
                        .Input => try daemon.handleInput(client, msg.payload),
                        .Send => daemon.handleSend(msg.payload),
                        .Output => try daemon.handleOutput(msg.payload, &vt_stream),
                        .Init => try daemon.handleInit(client, pty_fd, &term, msg.payload),
                        .Switch => try daemon.handleSwitch(msg.payload),

...	...		@@ -1046,6 +1046,11 @@ const Daemon = struct {
1046	1046		}
1047	1047		}
1048	1048
	1049	+	/// Queue input from `zmx send` without changing interactive client leadership.
	1050	+	pub fn handleSend(self: *Daemon, payload: []const u8) void {
	1051	+	self.queuePtyInput(payload);
	1052	+	}
	1053	+
1049	1054		pub fn handleSwitch(self: *Daemon, session_name: []const u8) !void {
1050	1055		for (self.clients.items) \|client\| {
1051	1056		if (self.leader_client_fd == client.socket_fd) {

...	...		@@ -1371,6 +1376,27 @@ const Daemon = struct {
1371	1376		}
1372	1377		};
1373	1378
	1379	+	test "send queues PTY input without changing leader" {
	1380	+	const alloc = std.testing.allocator;
	1381	+	var daemon = Daemon{
	1382	+	.cfg = undefined,
	1383	+	.alloc = alloc,
	1384	+	.clients = .empty,
	1385	+	.leader_client_fd = 42,
	1386	+	.session_name = "test",
	1387	+	.socket_path = "",
	1388	+	.running = true,
	1389	+	.pid = 0,
	1390	+	.created_at = 0,
	1391	+	};
	1392	+	defer daemon.pty_write_buf.deinit(alloc);
	1393	+
	1394	+	daemon.handleSend("hello");
	1395	+
	1396	+	try std.testing.expectEqual(@as(?i32, 42), daemon.leader_client_fd);
	1397	+	try std.testing.expectEqualStrings("hello", daemon.pty_write_buf.items);
	1398	+	}
	1399	+
1374	1400		fn printVersion(cfg: *Cfg) !void {
1375	1401		var buf: [256]u8 = undefined;
1376	1402		var w = std.fs.File.stdout().writer(&buf);

...	...		@@ -19,6 +19,7 @@ Use spec: https://common-changelog.org/
19	19		- `zmx run` will now detect heredocs and add the completion marker to a newline
20	20		- Race between `zmx kill X; zmx run X`
21	21		- Improved claim leader detection
	22	+	- `zmx send` no longer claims client leadership or triggers a resize probe
22	23		- Improved `ctrl+\` key detection
23	24		- Support for linux kernel < 4.11 by avoiding statx calls
24	25

...	...		@@ -22,15 +22,16 @@ pub const Tag = enum(u8) {
22	22		LabelSet = 15,
23	23		LabelClear = 16,
24	24		LabelData = 17,
	25	+	Send = 18,
25	26		// Non-exhaustive: this enum comes off the wire via bytesToValue and
26		-	// @enumFromInt, so out-of-range values (14-255) are representable
	27	+	// @enumFromInt, so out-of-range values are representable
27	28		// rather than UB. Switches must handle `_` (unknown tag).
28	29		_,
29	30		};
30	31
31	32		comptime {
32	33		if (@typeInfo(Tag).@"enum".is_exhaustive) @compileError(
33		-	"ipc.Tag must stay non-exhaustive — old daemons rely on `_` to ignore unknown tags",
	34	+	"ipc.Tag must stay non-exhaustive -- old daemons rely on `_` to ignore unknown tags",
34	35		);
35	36		}
36	37

...	...		@@ -57,7 +58,7 @@ pub fn getTerminalSize(fd: i32) Resize {
57	58		pub const MAX_CMD_LEN = 256;
58	59		pub const MAX_CWD_LEN = 256;
59	60
60		-	/// Frozen wire shape. Do NOT add fields — new stats go in new `Tag` values
	61	+	/// Frozen wire shape. Do NOT add fields! New stats go in new `Tag` values
61	62		/// so old daemons (whose `_` arm ignores unknown tags) stay reachable.
62	63		/// Changing `@sizeOf(Info)` breaks `zmx list` against running daemons.
63	64		pub const Info = extern struct {

...	...		@@ -306,6 +307,7 @@ test "Tag wire values are frozen" {
306	307		.{ Tag.Run, 9 }, .{ Tag.Ack, 10 }, .{ Tag.Switch, 11 },
307	308		.{ Tag.Write, 12 }, .{ Tag.TaskComplete, 13 }, .{ Tag.LabelGet, 14 },
308	309		.{ Tag.LabelSet, 15 }, .{ Tag.LabelClear, 16 }, .{ Tag.LabelData, 17 },
	310	+	.{ Tag.Send, 18 },
309	311		}) \|p\| try std.testing.expectEqual(@as(u8, p[1]), @intFromEnum(p[0]));
310	312		}
311	313

...	...		@@ -287,7 +287,7 @@ pub fn main() !void {
287	287		error.NameTooLong => return socket.printSessionNameTooLong(sesh, cfg.socket_dir),
288	288		error.OutOfMemory => return err,
289	289		};
290		-	return send(&cfg, sesh, socket_path, text_parts.items, .Input);
	290	+	return send(&cfg, sesh, socket_path, text_parts.items, .Send);
291	291		} else if (std.mem.eql(u8, cmd, "print") or std.mem.eql(u8, cmd, "p")) {
292	292		const session_name = args.next() orelse "";
293	293		if (std.mem.eql(u8, session_name, "--help") or std.mem.eql(u8, session_name, "-h")) {

...	...		@@ -3042,6 +3068,7 @@ fn daemonLoop(daemon: *Daemon, server_sock_fd: i32, pty_fd: i32) !void {
3042	3068		while (client.read_buf.next()) \|msg\| {
3043	3069		switch (msg.header.tag) {
3044	3070		.Input => try daemon.handleInput(client, msg.payload),
	3071	+	.Send => daemon.handleSend(msg.payload),
3045	3072		.Output => try daemon.handleOutput(msg.payload, &vt_stream),
3046	3073		.Init => try daemon.handleInit(client, pty_fd, &term, msg.payload),
3047	3074		.Switch => try daemon.handleSwitch(msg.payload),

Commit 8ba312d

Commit `8ba312d`