Eric Bower
·
2026-10-23
1const std = @import("std");
2const cross = @import("cross");
3const socket = @import("socket.zig");
4const lib_posix = @import("posix.zig");
5
6pub const PROTOCOL_VERSION: u8 = 1;
7
8pub const Tag = enum(u8) {
9 Input = 0,
10 Output = 1,
11 Resize = 2,
12 Detach = 3,
13 DetachAll = 4,
14 Kill = 5,
15 Info = 6,
16 Init = 7,
17 History = 8,
18 Run = 9,
19 Ack = 10,
20 Switch = 11,
21 Write = 12,
22 TaskComplete = 13,
23 LabelGet = 14,
24 LabelSet = 15,
25 LabelClear = 16,
26 LabelData = 17,
27 Send = 18,
28 EnvGet = 19,
29 EnvSet = 20,
30 EnvData = 21,
31 /// Scoped history: the active screen plus a bounded number of
32 /// preceding scrollback rows. Request payload is `Capture`.
33 Capture = 22,
34 Unsupported = 255,
35 // Non-exhaustive: this enum comes off the wire via bytesToValue and
36 // @enumFromInt, so out-of-range values are representable
37 // rather than UB. Switches must handle `_` (unknown tag).
38 _,
39};
40
41comptime {
42 if (@typeInfo(Tag).@"enum".mode == .exhaustive) @compileError(
43 "ipc.Tag must stay non-exhaustive -- old daemons rely on `_` to ignore unknown tags",
44 );
45}
46
47pub const Header = extern struct {
48 tag: Tag,
49 protocol_version: u8 = PROTOCOL_VERSION,
50 _reserved: u16 = 0,
51 len: u32,
52};
53
54pub const Resize = extern struct {
55 rows: u16,
56 cols: u16,
57 xpixel: u16 = 0,
58 ypixel: u16 = 0,
59
60 pub fn winsize(self: Resize) cross.struct_winsize {
61 return .{
62 .ws_row = self.rows,
63 .ws_col = self.cols,
64 .ws_xpixel = self.xpixel,
65 .ws_ypixel = self.ypixel,
66 };
67 }
68};
69
70/// Request payload for `Tag.Capture`. `format` is a `util.HistoryFormat`
71/// and `rows` is how many scrollback rows to include above the active
72/// screen; zero returns only the screen.
73pub const Capture = extern struct {
74 format: u8,
75 _reserved: [3]u8 = @splat(0),
76 rows: u32,
77};
78
79pub fn getTerminalSize(fd: i32) Resize {
80 var ws: cross.struct_winsize = undefined;
81 if (cross.ioctl(fd, cross.TIOCGWINSZ, &ws) == 0 and ws.ws_row > 0 and ws.ws_col > 0) {
82 return .{ .rows = ws.ws_row, .cols = ws.ws_col, .xpixel = ws.ws_xpixel, .ypixel = ws.ws_ypixel };
83 }
84 inline for (.{ lib_posix.STDOUT_FILENO, lib_posix.STDIN_FILENO, lib_posix.STDERR_FILENO }) |fallback_fd| {
85 if (fallback_fd != fd) {
86 if (cross.ioctl(fallback_fd, cross.TIOCGWINSZ, &ws) == 0 and ws.ws_row > 0 and ws.ws_col > 0) {
87 return .{ .rows = ws.ws_row, .cols = ws.ws_col, .xpixel = ws.ws_xpixel, .ypixel = ws.ws_ypixel };
88 }
89 }
90 }
91 if (lib_posix.open("/dev/tty", .{ .ACCMODE = .RDWR }, 0)) |tty_fd| {
92 defer lib_posix.close(tty_fd);
93 if (cross.ioctl(tty_fd, cross.TIOCGWINSZ, &ws) == 0 and ws.ws_row > 0 and ws.ws_col > 0) {
94 return .{ .rows = ws.ws_row, .cols = ws.ws_col, .xpixel = ws.ws_xpixel, .ypixel = ws.ws_ypixel };
95 }
96 } else |_| {}
97 return .{ .rows = 24, .cols = 120 };
98}
99
100pub const MAX_PAYLOAD_LEN: usize = 64 * 1024 * 1024;
101pub const MAX_CMD_LEN = 256;
102pub const MAX_CWD_LEN = 256;
103
104/// Frozen wire shape. Do NOT add fields! New stats go in new `Tag` values
105/// so old daemons (whose `_` arm ignores unknown tags) stay reachable.
106/// Changing `@sizeOf(Info)` breaks `zmx list` against running daemons.
107pub const Info = extern struct {
108 clients_len: u64,
109 pid: i32,
110 cmd_len: u16,
111 cwd_len: u16,
112 cmd: [MAX_CMD_LEN]u8,
113 cwd: [MAX_CWD_LEN]u8,
114 created_at: u64,
115 task_ended_at: u64,
116 task_exit_code: u8,
117 protocol_version: u8 = PROTOCOL_VERSION,
118 _reserved: [6]u8 = @splat(0),
119};
120
121pub fn send(fd: i32, tag: Tag, data: []const u8) !void {
122 const header = Header{
123 .tag = tag,
124 .len = @intCast(data.len),
125 };
126 const header_bytes = std.mem.asBytes(&header);
127 try writeAll(fd, header_bytes);
128 if (data.len > 0) {
129 try writeAll(fd, data);
130 }
131}
132
133pub fn appendMessage(
134 gpa: std.mem.Allocator,
135 list: *std.ArrayList(u8),
136 tag: Tag,
137 data: []const u8,
138) !void {
139 const header = Header{
140 .tag = tag,
141 .len = @intCast(data.len),
142 };
143 // Guarantee capacity for header + payload in one check to avoid
144 // intermediate realloc between the two appends on the hot path.
145 try list.ensureTotalCapacity(gpa, list.items.len + @sizeOf(Header) + data.len);
146 list.appendSliceAssumeCapacity(std.mem.asBytes(&header));
147 if (data.len > 0) {
148 list.appendSliceAssumeCapacity(data);
149 }
150}
151
152fn writeAll(fd: i32, data: []const u8) !void {
153 var index: usize = 0;
154 while (index < data.len) {
155 const n = try lib_posix.write(fd, data[index..]);
156 if (n == 0) return error.DiskQuota;
157 index += n;
158 }
159}
160
161pub const Message = struct {
162 tag: Tag,
163 data: []u8,
164
165 pub fn deinit(self: Message, alloc: std.mem.Allocator) void {
166 if (self.data.len > 0) {
167 alloc.free(self.data);
168 }
169 }
170};
171
172pub const SocketMsg = struct {
173 header: Header,
174 payload: []const u8,
175};
176
177pub const SocketBuffer = struct {
178 buf: std.ArrayList(u8),
179 alloc: std.mem.Allocator,
180 head: usize,
181
182 pub fn init(alloc: std.mem.Allocator) !SocketBuffer {
183 return .{
184 .buf = try std.ArrayList(u8).initCapacity(alloc, 4096),
185 .alloc = alloc,
186 .head = 0,
187 };
188 }
189
190 pub fn deinit(self: *SocketBuffer) void {
191 self.buf.deinit(self.alloc);
192 }
193
194 /// Reads from fd into buffer.
195 /// Returns number of bytes read.
196 /// Propagates error.WouldBlock and other errors to caller.
197 /// Returns 0 on EOF.
198 pub fn read(self: *SocketBuffer, fd: i32) !usize {
199 if (self.head > 0) {
200 const remaining = self.buf.items.len - self.head;
201 if (remaining > 0) {
202 std.mem.copyForwards(u8, self.buf.items[0..remaining], self.buf.items[self.head..]);
203 self.buf.items.len = remaining;
204 } else {
205 self.buf.clearRetainingCapacity();
206 }
207 self.head = 0;
208 }
209
210 var tmp: [4096]u8 = undefined;
211 const n = try lib_posix.read(fd, &tmp);
212 if (n > 0) {
213 try self.buf.appendSlice(self.alloc, tmp[0..n]);
214 }
215 return n;
216 }
217
218 /// Returns the next complete message or `null` when none available.
219 /// `buf` is advanced automatically; caller keeps the returned slices
220 /// valid until the following `next()` (or `deinit`).
221 pub fn next(self: *SocketBuffer) ?SocketMsg {
222 const available = self.buf.items[self.head..];
223 if (available.len < @sizeOf(Header)) return null;
224
225 const hdr = std.mem.bytesToValue(Header, available[0..@sizeOf(Header)]);
226 if (hdr.len > MAX_PAYLOAD_LEN or hdr.protocol_version != PROTOCOL_VERSION) {
227 // framing error / corrupted stream: clear buffer to avoid
228 // infinite alloc
229 self.buf.clearRetainingCapacity();
230 self.head = 0;
231 return null;
232 }
233
234 const total = @sizeOf(Header) + @as(usize, hdr.len);
235 if (available.len < total) return null;
236
237 const pay = available[@sizeOf(Header)..total];
238 self.head += total;
239 return .{ .header = hdr, .payload = pay };
240 }
241};
242
243const ConnectError = error{
244 ConnectionRefused,
245 Unexpected,
246};
247
248/// Connect-only liveness check. Callers that don't read `Info` should use
249/// this (not `probeSession`) so they survive `Info` shape changes.
250pub fn connectSession(socket_path: []const u8) ConnectError!i32 {
251 return socket.sessionConnect(socket_path) catch |err| switch (err) {
252 error.ConnectionRefused => return error.ConnectionRefused,
253 else => return error.Unexpected,
254 };
255}
256
257const SessionProbeError = error{
258 Timeout,
259 ConnectionRefused,
260 Unexpected,
261 InfoSizeMismatch,
262 Unsupported,
263};
264
265const SessionProbeResult = struct {
266 fd: i32,
267 info: Info,
268 labels: ?[]const u8,
269 alloc: std.mem.Allocator,
270
271 pub fn deinit(self: *const SessionProbeResult) void {
272 if (self.labels) |lbl| self.alloc.free(lbl);
273 lib_posix.close(self.fd);
274 }
275};
276
277pub fn probeSession(
278 alloc: std.mem.Allocator,
279 socket_path: []const u8,
280) SessionProbeError!SessionProbeResult {
281 const timeout_ms = 1000;
282 const fd = try connectSession(socket_path);
283 errdefer lib_posix.close(fd);
284
285 send(fd, .Info, "") catch return error.Unexpected;
286 send(fd, .LabelGet, "") catch {};
287
288 var poll_fds = [_]lib_posix.pollfd{.{ .fd = fd, .events = lib_posix.POLL.IN, .revents = 0 }};
289 const poll_result = lib_posix.poll(&poll_fds, timeout_ms) catch return error.Unexpected;
290 if (poll_result == 0) {
291 return error.Timeout;
292 }
293
294 var sb = SocketBuffer.init(alloc) catch return error.Unexpected;
295 defer sb.deinit();
296
297 const n = sb.read(fd) catch return error.Unexpected;
298 if (n == 0) return error.Unexpected;
299
300 var info_result: ?Info = null;
301 var labels: ?[]const u8 = null;
302 errdefer if (labels) |lbl| alloc.free(lbl);
303
304 while (true) {
305 if (sb.next()) |msg| {
306 if (msg.header.tag == .Info) {
307 if (msg.payload.len != @sizeOf(Info)) return error.InfoSizeMismatch;
308 info_result = std.mem.bytesToValue(Info, msg.payload[0..@sizeOf(Info)]);
309 }
310 if (msg.header.tag == .LabelData) {
311 labels = alloc.dupe(u8, msg.payload) catch null;
312 }
313
314 if (info_result != null and labels != null) break;
315 continue;
316 }
317
318 // No complete message available, wait for more data
319 const more = lib_posix.poll(&poll_fds, 50) catch break;
320 if (more == 0) break;
321 const n_read = sb.read(fd) catch break;
322 if (n_read == 0) break;
323 }
324
325 if (info_result) |info| {
326 return .{
327 .fd = fd,
328 .info = info,
329 .labels = labels,
330 .alloc = alloc,
331 };
332 }
333 return error.Unexpected;
334}
335
336// WIRE PROTOCOL FREEZE: read before "fixing" any test below.
337//
338// Changing these constants does not fix the test; it breaks every
339// running daemon for every user until they `pkill -f zmx`.
340//
341// Need a new field? → add a new `Tag` value (next free integer).
342// Need to remove one? → don't. Reserve the integer, stop sending it.
343test "Info wire size is frozen" {
344 try std.testing.expectEqual(@as(usize, 552), @sizeOf(Info));
345 try std.testing.expectEqual(@as(usize, 8), @sizeOf(Header));
346 try std.testing.expectEqual(@as(usize, 8), @sizeOf(Resize));
347 try std.testing.expectEqual(@as(usize, 8), @sizeOf(Capture));
348}
349
350test "Tag wire values are frozen" {
351 inline for (.{
352 .{ Tag.Input, 0 }, .{ Tag.Output, 1 }, .{ Tag.Resize, 2 },
353 .{ Tag.Detach, 3 }, .{ Tag.DetachAll, 4 }, .{ Tag.Kill, 5 },
354 .{ Tag.Info, 6 }, .{ Tag.Init, 7 }, .{ Tag.History, 8 },
355 .{ Tag.Run, 9 }, .{ Tag.Ack, 10 }, .{ Tag.Switch, 11 },
356 .{ Tag.Write, 12 }, .{ Tag.TaskComplete, 13 }, .{ Tag.LabelGet, 14 },
357 .{ Tag.LabelSet, 15 }, .{ Tag.LabelClear, 16 }, .{ Tag.LabelData, 17 },
358 .{ Tag.Send, 18 }, .{ Tag.EnvGet, 19 }, .{ Tag.EnvSet, 20 },
359 .{ Tag.EnvData, 21 }, .{ Tag.Capture, 22 }, .{ Tag.Unsupported, 255 },
360 }) |p| try std.testing.expectEqual(@as(u8, p[1]), @backingInt(p[0]));
361}
362
363pub fn roundTripForTag(
364 alloc: std.mem.Allocator,
365 socket_path: []const u8,
366 request_tag: Tag,
367 payload: []const u8,
368 expected_tag: Tag,
369) SessionProbeError![]u8 {
370 const timeout_ms = 1000;
371 const fd = try connectSession(socket_path);
372 defer lib_posix.close(fd);
373
374 send(fd, request_tag, payload) catch return error.Unexpected;
375
376 var poll_fds = [_]lib_posix.pollfd{.{ .fd = fd, .events = lib_posix.POLL.IN, .revents = 0 }};
377 const poll_result = lib_posix.poll(&poll_fds, timeout_ms) catch return error.Unexpected;
378 if (poll_result == 0) return error.Timeout;
379
380 var sb = SocketBuffer.init(alloc) catch return error.Unexpected;
381 defer sb.deinit();
382
383 const n = sb.read(fd) catch return error.Unexpected;
384 if (n == 0) return error.Unexpected;
385
386 while (sb.next()) |msg| {
387 if (msg.header.tag == .Unsupported) return error.Unsupported;
388 if (msg.header.tag == expected_tag) {
389 return alloc.dupe(u8, msg.payload) catch return error.Unexpected;
390 }
391 }
392 return error.Unexpected;
393}
394
395test "zeroed Info has no stack garbage in wire bytes" {
396 var info = std.mem.zeroes(Info);
397 info.clients_len = 3;
398 info.pid = 999;
399 info.task_exit_code = 7;
400 const bytes = std.mem.asBytes(&info);
401 // Tail padding after task_exit_code must be zero (asBytes ships it).
402 const last_field_end = @offsetOf(Info, "task_exit_code") + @sizeOf(u8);
403 for (bytes[last_field_end..]) |b| try std.testing.expectEqual(@as(u8, 0), b);
404}