Eric Bower
·
2026-10-23
1//! libzmx - programmatic client interface for zmx IPC.
2const std = @import("std");
3const Allocator = std.mem.Allocator;
4const Io = std.Io;
5const build_options = @import("build_options");
6
7pub const has_daemon = if (@hasDecl(build_options, "has_daemon")) build_options.has_daemon else true;
8
9const ipc = @import("ipc.zig");
10const socket = @import("socket.zig");
11const Cfg = @import("cfg.zig");
12const util = @import("util.zig");
13const label = @import("label.zig");
14const lib_posix = @import("posix.zig");
15const cross = @import("cross");
16const signal = @import("signal.zig");
17const loop = @import("loop.zig");
18const Daemon = if (has_daemon) loop.Daemon else struct {};
19
20pub const HistoryFormat = util.HistoryFormat;
21pub const SessionEntry = util.SessionEntry;
22
23pub const CreateOptions = struct {
24 command: ?[]const []const u8 = null,
25 cwd: ?[]const u8 = null,
26 shell: ?[]const u8 = null,
27 labels: ?[]const u8 = null,
28 is_task_mode: bool = false,
29 quiet: bool = false,
30};
31
32pub const AttachOptions = struct {
33 command: ?[]const []const u8 = null,
34 cwd: ?[]const u8 = null,
35 shell: ?[]const u8 = null,
36 labels: ?[]const u8 = null,
37 tracked_env: ?[]const u8 = null,
38};
39
40pub const RunOptions = struct {
41 detached: bool = false,
42 cwd: ?[]const u8 = null,
43 shell: ?[]const u8 = null,
44 stdin_bytes: ?[]const u8 = null,
45 output_writer: ?*Io.Writer = null,
46};
47
48pub const KillOptions = struct {
49 force: bool = false,
50};
51
52/// Ensure a session daemon is running in the background without attaching.
53/// Resolves socket_dir and session prefix automatically.
54/// If the session does not exist yet, forks the daemon internally (and exits
55/// the daemon child process when the session ends so it never returns to caller).
56/// Returns true if a new session was created, false if it already existed.
57pub fn create(alloc: Allocator, io: Io, name: []const u8, opts: CreateOptions) !bool {
58 if (!has_daemon) @compileError("create() requires -Ddaemon=true");
59 signal.ignoreSigpipe();
60 const sesh = try socket.getSeshName(alloc, name);
61 defer alloc.free(sesh);
62
63 var cfg = try Cfg.init(alloc, io);
64 defer cfg.deinit(alloc);
65
66 const socket_path = try socket.getSocketPath(alloc, cfg.socket_dir, sesh);
67 defer alloc.free(socket_path);
68
69 var dir = try std.Io.Dir.openDirAbsolute(io, cfg.socket_dir, .{});
70 defer dir.close(io);
71
72 const exists = try socket.sessionExists(io, dir, sesh);
73 var should_create = !exists;
74 if (exists) {
75 if (ipc.connectSession(socket_path)) |fd| {
76 lib_posix.close(fd);
77 return false;
78 } else |err| switch (err) {
79 error.ConnectionRefused => {
80 socket.cleanupStaleSocket(io, dir, sesh);
81 should_create = true;
82 },
83 else => return false,
84 }
85 }
86
87 if (!should_create) return false;
88
89 var daemon = Daemon.init(io, &cfg, sesh, socket_path);
90 daemon.command = opts.command;
91 if (opts.cwd) |cwd| daemon.setCwd(cwd);
92 if (opts.shell) |sh| daemon.shell = sh;
93 daemon.is_task_mode = opts.is_task_mode;
94 daemon.quiet = opts.quiet;
95
96 const is_daemon_proc = try daemon.ensureSession(io);
97 if (is_daemon_proc) {
98 std.process.exit(0);
99 }
100
101 if (opts.labels) |kvs| {
102 try set(alloc, io, sesh, kvs);
103 }
104
105 return true;
106}
107
108/// Connect to an existing session's Unix domain socket and return the socket fd.
109pub fn connect(alloc: Allocator, io: Io, name: []const u8) !i32 {
110 const sesh = try socket.getSeshName(alloc, name);
111 defer alloc.free(sesh);
112
113 var cfg = try Cfg.init(alloc, io);
114 defer cfg.deinit(alloc);
115
116 const socket_path = try socket.getSocketPath(alloc, cfg.socket_dir, sesh);
117 defer alloc.free(socket_path);
118
119 return socket.sessionConnect(socket_path);
120}
121
122/// Attach the current process's stdin/stdout TTY to a session (creating if needed).
123pub fn attach(alloc: Allocator, io: Io, name: []const u8, opts: AttachOptions) !void {
124 signal.ignoreSigpipe();
125 const cur_sesh = socket.getSeshNameFromEnv();
126 if (cur_sesh.len > 0) {
127 var cfg = try Cfg.init(alloc, io);
128 defer cfg.deinit(alloc);
129 const socket_path = try socket.getSocketPath(alloc, cfg.socket_dir, cur_sesh);
130 defer alloc.free(socket_path);
131 const fd = try socket.sessionConnect(socket_path);
132 defer lib_posix.close(fd);
133 const target_sesh = try socket.getSeshName(alloc, name);
134 defer alloc.free(target_sesh);
135 ipc.send(fd, .Switch, target_sesh) catch |err| switch (err) {
136 error.BrokenPipe, error.ConnectionResetByPeer => return,
137 else => return err,
138 };
139 return;
140 }
141
142 const sesh = try socket.getSeshName(alloc, name);
143 defer alloc.free(sesh);
144
145 var cfg = try Cfg.init(alloc, io);
146 defer cfg.deinit(alloc);
147
148 const socket_path = try socket.getSocketPath(alloc, cfg.socket_dir, sesh);
149 defer alloc.free(socket_path);
150
151 if (has_daemon) {
152 var daemon = Daemon.init(io, &cfg, sesh, socket_path);
153 daemon.command = opts.command;
154 if (opts.cwd) |cwd| daemon.setCwd(cwd);
155 if (opts.shell) |sh| daemon.shell = sh;
156 const is_daemon_proc = try daemon.ensureSession(io);
157 if (is_daemon_proc) std.process.exit(0);
158 }
159
160 if (opts.labels) |kvs| {
161 try set(alloc, io, sesh, kvs);
162 }
163
164 const client_sock = try socket.sessionConnect(socket_path);
165
166 var orig_termios: cross.termios = undefined;
167 const stdin_is_tty = cross.tcgetattr(lib_posix.STDIN_FILENO, &orig_termios) == 0;
168 const restore_seq = "\x1bc\x1b]110\x1b\\\x1b]111\x1b\\\x1b]112\x1b\\";
169
170 defer {
171 if (stdin_is_tty) {
172 _ = cross.tcsetattr(lib_posix.STDIN_FILENO, cross.TCSAFLUSH, &orig_termios);
173 }
174 _ = lib_posix.write(lib_posix.STDOUT_FILENO, restore_seq) catch {};
175 }
176
177 if (stdin_is_tty) {
178 var raw_termios = orig_termios;
179 cross.cfmakeraw(&raw_termios);
180 raw_termios.c_cc[cross.VLNEXT] = cross._POSIX_VDISABLE;
181 raw_termios.c_cc[cross.VQUIT] = cross._POSIX_VDISABLE;
182 raw_termios.c_cc[cross.VMIN] = 1;
183 raw_termios.c_cc[cross.VTIME] = 0;
184 _ = cross.tcsetattr(lib_posix.STDIN_FILENO, cross.TCSANOW, &raw_termios);
185 }
186
187 const clear_seq = "\x1b[0m\x1b[2J\x1b[H";
188 _ = try lib_posix.write(lib_posix.STDOUT_FILENO, clear_seq);
189
190 const env_str = opts.tracked_env orelse "";
191 const looper = try loop.clientLoop(client_sock, env_str);
192 switch (looper.kind) {
193 .detach => return,
194 .switch_session => {
195 if (looper.session_name) |target_name| {
196 _ = lib_posix.write(lib_posix.STDOUT_FILENO, restore_seq) catch {};
197 var new_opts = opts;
198 if (looper.cwd) |new_cwd| new_opts.cwd = new_cwd;
199 return attach(alloc, io, target_name, new_opts);
200 }
201 },
202 }
203}
204
205/// Detach all clients from the specified session (or current ZMX_SESSION if "." or empty).
206pub fn detach(alloc: Allocator, io: Io, name: []const u8) !void {
207 signal.ignoreSigpipe();
208 const raw_name = if (name.len == 0 or std.mem.eql(u8, name, "."))
209 socket.getSeshNameFromEnv()
210 else
211 name;
212 if (raw_name.len == 0) return error.SessionNameRequired;
213
214 const sesh = try socket.getSeshName(alloc, raw_name);
215 defer alloc.free(sesh);
216
217 var cfg = try Cfg.init(alloc, io);
218 defer cfg.deinit(alloc);
219
220 const socket_path = try socket.getSocketPath(alloc, cfg.socket_dir, sesh);
221 defer alloc.free(socket_path);
222
223 var dir = try std.Io.Dir.openDirAbsolute(io, cfg.socket_dir, .{});
224 defer dir.close(io);
225
226 const fd = ipc.connectSession(socket_path) catch |err| {
227 if (err == error.ConnectionRefused) socket.cleanupStaleSocket(io, dir, sesh);
228 return error.SessionUnresponsive;
229 };
230 defer lib_posix.close(fd);
231
232 ipc.send(fd, .DetachAll, "") catch |err| switch (err) {
233 error.BrokenPipe, error.ConnectionResetByPeer => return,
234 else => return err,
235 };
236}
237
238/// List all active sessions. Caller owns the returned slice and entries.
239pub fn list(alloc: Allocator, io: Io) ![]SessionEntry {
240 var cfg = try Cfg.init(alloc, io);
241 defer cfg.deinit(alloc);
242
243 var sessions = try util.get_session_entries(alloc, io, cfg.socket_dir);
244 return sessions.toOwnedSlice(alloc);
245}
246
247pub const KillStatus = enum {
248 killed,
249 cleaned_stale,
250};
251
252/// Kill a session and wait for its socket to close.
253pub fn kill(alloc: Allocator, io: Io, name: []const u8, opts: KillOptions) !KillStatus {
254 signal.ignoreSigpipe();
255 const sesh = try socket.getSeshName(alloc, name);
256 defer alloc.free(sesh);
257
258 var cfg = try Cfg.init(alloc, io);
259 defer cfg.deinit(alloc);
260
261 const socket_path = try socket.getSocketPath(alloc, cfg.socket_dir, sesh);
262 defer alloc.free(socket_path);
263
264 var dir = try std.Io.Dir.openDirAbsolute(io, cfg.socket_dir, .{});
265 defer dir.close(io);
266
267 const exists = try socket.sessionExists(io, dir, sesh);
268 if (!exists) return error.SessionNotFound;
269
270 const fd = ipc.connectSession(socket_path) catch |err| {
271 if (opts.force or err == error.ConnectionRefused) {
272 socket.cleanupStaleSocket(io, dir, sesh);
273 return .cleaned_stale;
274 }
275 return error.SessionUnresponsive;
276 };
277 defer lib_posix.close(fd);
278
279 ipc.send(fd, .Kill, "") catch |err| switch (err) {
280 error.BrokenPipe, error.ConnectionResetByPeer => return .killed,
281 else => return err,
282 };
283
284 var drain: [256]u8 = undefined;
285 while (true) {
286 const n = lib_posix.read(fd, &drain) catch break;
287 if (n == 0) break;
288 }
289 return .killed;
290}
291
292fn tailInternal(
293 alloc: Allocator,
294 client_socket_fds: []const i32,
295 detached: bool,
296 is_run_cmd: bool,
297 writer: ?*Io.Writer,
298) !u8 {
299 var poll_fds = try std.ArrayList(lib_posix.pollfd).initCapacity(alloc, client_socket_fds.len + 1);
300 defer poll_fds.deinit(alloc);
301
302 var read_buf = try ipc.SocketBuffer.init(alloc);
303 defer read_buf.deinit();
304
305 var out_buf = try std.ArrayList(u8).initCapacity(alloc, 4096);
306 defer out_buf.deinit(alloc);
307
308 var is_first_line = true;
309 var task_complete_code: ?u8 = null;
310
311 while (true) {
312 poll_fds.clearRetainingCapacity();
313
314 for (client_socket_fds) |fd| {
315 try poll_fds.append(alloc, .{
316 .fd = fd,
317 .events = lib_posix.POLL.IN,
318 .revents = 0,
319 });
320 }
321
322 if (out_buf.items.len > 0 and writer == null) {
323 try poll_fds.append(alloc, .{
324 .fd = lib_posix.STDOUT_FILENO,
325 .events = lib_posix.POLL.OUT,
326 .revents = 0,
327 });
328 }
329
330 _ = lib_posix.poll(poll_fds.items, -1) catch |err| {
331 if (err == error.Interrupted) continue;
332 return err;
333 };
334
335 for (poll_fds.items) |*poll_fd| {
336 if (poll_fd.revents & lib_posix.POLL.IN != 0) {
337 const n = read_buf.read(poll_fd.fd) catch |err| {
338 if (err == error.WouldBlock) continue;
339 if (err == error.ConnectionResetByPeer or err == error.BrokenPipe) return 1;
340 return err;
341 };
342 if (n == 0) {
343 if (task_complete_code) |exit_code| return exit_code;
344 return 0;
345 }
346
347 while (read_buf.next()) |msg| {
348 switch (msg.header.tag) {
349 .Ack => {
350 if (detached) {
351 if (writer) |w| {
352 try w.writeAll("command sent!\n");
353 } else {
354 _ = lib_posix.write(lib_posix.STDOUT_FILENO, "command sent!\n") catch {};
355 }
356 return 0;
357 }
358 },
359 .Output => {
360 if (msg.payload.len > 0) {
361 var payload = msg.payload;
362 if (!detached and is_run_cmd and is_first_line) {
363 if (std.mem.indexOfScalar(u8, payload, '\n')) |nl| {
364 is_first_line = false;
365 payload = payload[nl + 1 ..];
366 } else {
367 is_first_line = false;
368 payload = payload[payload.len..];
369 }
370 }
371
372 if (payload.len > 0) {
373 const plain = util.stripAnsi(alloc, payload) catch continue;
374 defer alloc.free(plain);
375 if (plain.len > 0) {
376 if (writer) |w| {
377 try w.writeAll(plain);
378 } else {
379 try out_buf.appendSlice(alloc, plain);
380 }
381 }
382 }
383 }
384 },
385 .TaskComplete => {
386 task_complete_code = if (msg.payload.len > 0) msg.payload[0] else 0;
387 },
388 else => {},
389 }
390 }
391 }
392 }
393
394 if (task_complete_code) |exit_code| {
395 if (writer == null) {
396 while (out_buf.items.len > 0) {
397 const n = lib_posix.write(lib_posix.STDOUT_FILENO, out_buf.items) catch break;
398 try out_buf.replaceRange(alloc, 0, n, &[_]u8{});
399 }
400 }
401 return exit_code;
402 }
403
404 if (writer == null and out_buf.items.len > 0) {
405 const n = lib_posix.write(lib_posix.STDOUT_FILENO, out_buf.items) catch |err| blk: {
406 if (err == error.WouldBlock) break :blk 0;
407 return err;
408 };
409 if (n > 0) {
410 try out_buf.replaceRange(alloc, 0, n, &[_]u8{});
411 }
412 }
413
414 for (poll_fds.items) |poll_fd| {
415 if (poll_fd.revents & (lib_posix.POLL.HUP | lib_posix.POLL.ERR | lib_posix.POLL.NVAL) != 0) {
416 return 0;
417 }
418 }
419 }
420}
421
422/// Run a command inside a session in task mode (creating if needed).
423/// Returns the command's exit code (or 0 immediately if `opts.detached` is true).
424pub fn run(
425 alloc: Allocator,
426 io: Io,
427 name: []const u8,
428 command: []const []const u8,
429 opts: RunOptions,
430) !u8 {
431 signal.ignoreSigpipe();
432 const sesh = try socket.getSeshName(alloc, name);
433 defer alloc.free(sesh);
434
435 var cfg = try Cfg.init(alloc, io);
436 defer cfg.deinit(alloc);
437
438 const socket_path = try socket.getSocketPath(alloc, cfg.socket_dir, sesh);
439 defer alloc.free(socket_path);
440
441 if (has_daemon) {
442 var daemon = Daemon.init(io, &cfg, sesh, socket_path);
443 daemon.is_task_mode = true;
444 if (opts.cwd) |cwd| daemon.setCwd(cwd);
445 if (opts.shell) |sh| daemon.shell = sh;
446 const is_daemon_proc = try daemon.ensureSession(io);
447 if (is_daemon_proc) std.process.exit(0);
448 }
449
450 var cmd_to_send: ?[]const u8 = null;
451 var allocated_cmd: ?[]u8 = null;
452 defer if (allocated_cmd) |c| alloc.free(c);
453
454 if (command.len > 0) {
455 var cmd_list = std.ArrayList(u8).empty;
456 defer cmd_list.deinit(alloc);
457
458 for (command, 0..) |arg, i| {
459 if (i > 0) try cmd_list.append(alloc, ' ');
460 if (util.shellNeedsQuoting(arg)) {
461 const quoted = try util.shellQuote(alloc, arg);
462 defer alloc.free(quoted);
463 try cmd_list.appendSlice(alloc, quoted);
464 } else {
465 try cmd_list.appendSlice(alloc, arg);
466 }
467 }
468 try cmd_list.append(alloc, '\r');
469 cmd_to_send = try cmd_list.toOwnedSlice(alloc);
470 allocated_cmd = @constCast(cmd_to_send.?);
471 } else if (opts.stdin_bytes) |bytes| {
472 if (bytes.len > 0) {
473 var buf = try std.ArrayList(u8).initCapacity(alloc, bytes.len + 1);
474 defer buf.deinit(alloc);
475 try buf.appendSlice(alloc, bytes);
476 if (buf.items[buf.items.len - 1] == '\n') {
477 buf.items[buf.items.len - 1] = '\r';
478 } else {
479 try buf.append(alloc, '\r');
480 }
481 cmd_to_send = try buf.toOwnedSlice(alloc);
482 allocated_cmd = @constCast(cmd_to_send.?);
483 }
484 }
485
486 if (cmd_to_send == null) return error.CommandRequired;
487
488 const client_sock = try socket.sessionConnect(socket_path);
489 defer lib_posix.close(client_sock);
490
491 const term_size = ipc.getTerminalSize(lib_posix.STDOUT_FILENO);
492 ipc.send(client_sock, .Resize, std.mem.asBytes(&term_size)) catch {};
493
494 ipc.send(client_sock, .Run, cmd_to_send.?) catch |err| switch (err) {
495 error.ConnectionResetByPeer, error.BrokenPipe => return 0,
496 else => return err,
497 };
498
499 const fds = [_]i32{client_sock};
500 return tailInternal(alloc, &fds, opts.detached, true, opts.output_writer);
501}
502
503/// Wait for one or more task-mode sessions to complete and return aggregate exit code.
504pub fn wait(alloc: Allocator, io: Io, names: []const []const u8) !u8 {
505 if (names.len == 0) return error.SessionNameRequired;
506 var cfg = try Cfg.init(alloc, io);
507 defer cfg.deinit(alloc);
508
509 var matchers = try std.ArrayList(socket.SessionMatch).initCapacity(alloc, names.len);
510 defer {
511 for (matchers.items) |m| alloc.free(m.name);
512 matchers.deinit(alloc);
513 }
514 for (names) |name| {
515 try matchers.append(alloc, try socket.parseSessionArg(alloc, name));
516 }
517
518 var agg_exit_code: u8 = 0;
519 while (true) {
520 agg_exit_code = 0;
521 var sessions = try util.get_session_entries(alloc, io, cfg.socket_dir);
522 defer {
523 for (sessions.items) |s| s.deinit(alloc);
524 sessions.deinit(alloc);
525 }
526
527 var total: usize = 0;
528 var done: usize = 0;
529
530 for (sessions.items) |session| {
531 var matched = false;
532 for (matchers.items) |m| {
533 if (m.matches(session.name)) {
534 matched = true;
535 break;
536 }
537 }
538 if (!matched) continue;
539 total += 1;
540 if (session.is_error) {
541 agg_exit_code = 1;
542 done += 1;
543 } else if (session.task_ended_at != null) {
544 if (session.task_exit_code) |ec| {
545 if (ec != 0) agg_exit_code = ec;
546 }
547 done += 1;
548 }
549 }
550
551 if (total > 0 and done == total) return agg_exit_code;
552 std.time.sleep(50 * std.time.ns_per_ms);
553 }
554}
555
556/// Follow live output from one or more sessions, writing plain text to `writer`.
557pub fn tail(alloc: Allocator, io: Io, names: []const []const u8, writer: ?*Io.Writer) !u8 {
558 if (names.len == 0) return error.SessionNameRequired;
559 var cfg = try Cfg.init(alloc, io);
560 defer cfg.deinit(alloc);
561
562 var matchers = try std.ArrayList(socket.SessionMatch).initCapacity(alloc, names.len);
563 defer {
564 for (matchers.items) |m| alloc.free(m.name);
565 matchers.deinit(alloc);
566 }
567 for (names) |name| {
568 try matchers.append(alloc, try socket.parseSessionArg(alloc, name));
569 }
570
571 var resolved_names: std.ArrayList([]const u8) = .empty;
572 defer {
573 for (resolved_names.items) |n| alloc.free(n);
574 resolved_names.deinit(alloc);
575 }
576
577 var any_prefix = false;
578 for (matchers.items) |m| {
579 if (m.is_prefix) {
580 any_prefix = true;
581 break;
582 }
583 }
584
585 if (any_prefix) {
586 var sessions = try util.get_session_entries(alloc, io, cfg.socket_dir);
587 defer {
588 for (sessions.items) |s| s.deinit(alloc);
589 sessions.deinit(alloc);
590 }
591 for (sessions.items) |s| {
592 for (matchers.items) |m| {
593 if (m.matches(s.name)) {
594 try resolved_names.append(alloc, try alloc.dupe(u8, s.name));
595 break;
596 }
597 }
598 }
599 }
600
601 for (matchers.items) |m| {
602 if (!m.is_prefix) {
603 try resolved_names.append(alloc, try alloc.dupe(u8, m.name));
604 }
605 }
606
607 if (resolved_names.items.len == 0) return error.SessionNotFound;
608
609 var fds = try std.ArrayList(i32).initCapacity(alloc, resolved_names.items.len);
610 defer {
611 for (fds.items) |fd| lib_posix.close(fd);
612 fds.deinit(alloc);
613 }
614
615 for (resolved_names.items) |session_name| {
616 const socket_path = try socket.getSocketPath(alloc, cfg.socket_dir, session_name);
617 defer alloc.free(socket_path);
618 const sock = try socket.sessionConnect(socket_path);
619 try fds.append(alloc, sock);
620 }
621
622 return tailInternal(alloc, fds.items, false, false, writer);
623}
624
625/// Send raw input bytes to a session's PTY (fire-and-forget, like `zmx send`).
626pub fn send(alloc: Allocator, io: Io, name: []const u8, bytes: []const u8) !void {
627 signal.ignoreSigpipe();
628 const sesh = try socket.getSeshName(alloc, name);
629 defer alloc.free(sesh);
630
631 var cfg = try Cfg.init(alloc, io);
632 defer cfg.deinit(alloc);
633
634 const socket_path = try socket.getSocketPath(alloc, cfg.socket_dir, sesh);
635 defer alloc.free(socket_path);
636
637 var dir = try std.Io.Dir.openDirAbsolute(io, cfg.socket_dir, .{});
638 defer dir.close(io);
639
640 const probe_result = ipc.probeSession(alloc, socket_path) catch |err| {
641 if (err == error.ConnectionRefused) socket.cleanupStaleSocket(io, dir, sesh);
642 return error.SessionUnresponsive;
643 };
644 defer probe_result.deinit();
645
646 ipc.send(probe_result.fd, .Send, bytes) catch |err| switch (err) {
647 error.ConnectionResetByPeer, error.BrokenPipe => return,
648 else => return err,
649 };
650}
651
652/// Inject raw bytes directly into a session's display/scrollback (like `zmx print`).
653pub fn print(alloc: Allocator, io: Io, name: []const u8, bytes: []const u8) !void {
654 signal.ignoreSigpipe();
655 const sesh = try socket.getSeshName(alloc, name);
656 defer alloc.free(sesh);
657
658 var cfg = try Cfg.init(alloc, io);
659 defer cfg.deinit(alloc);
660
661 const socket_path = try socket.getSocketPath(alloc, cfg.socket_dir, sesh);
662 defer alloc.free(socket_path);
663
664 var dir = try std.Io.Dir.openDirAbsolute(io, cfg.socket_dir, .{});
665 defer dir.close(io);
666
667 const probe_result = ipc.probeSession(alloc, socket_path) catch |err| {
668 if (err == error.ConnectionRefused) socket.cleanupStaleSocket(io, dir, sesh);
669 return error.SessionUnresponsive;
670 };
671 defer probe_result.deinit();
672
673 ipc.send(probe_result.fd, .Output, bytes) catch |err| switch (err) {
674 error.ConnectionResetByPeer, error.BrokenPipe => return,
675 else => return err,
676 };
677}
678
679/// Write `content` to `remote_path` inside the session (like `zmx write`).
680pub fn write(alloc: Allocator, io: Io, name: []const u8, remote_path: []const u8, content: []const u8) !void {
681 signal.ignoreSigpipe();
682 const sesh = try socket.getSeshName(alloc, name);
683 defer alloc.free(sesh);
684
685 var cfg = try Cfg.init(alloc, io);
686 defer cfg.deinit(alloc);
687
688 const socket_path = try socket.getSocketPath(alloc, cfg.socket_dir, sesh);
689 defer alloc.free(socket_path);
690
691 var dir = try std.Io.Dir.openDirAbsolute(io, cfg.socket_dir, .{});
692 defer dir.close(io);
693
694 const result = ipc.probeSession(alloc, socket_path) catch |err| {
695 if (err == error.ConnectionRefused) socket.cleanupStaleSocket(io, dir, sesh);
696 return error.SessionUnresponsive;
697 };
698 defer result.deinit();
699
700 var wire_buf = try std.ArrayList(u8).initCapacity(alloc, @sizeOf(u32) + remote_path.len + content.len);
701 defer wire_buf.deinit(alloc);
702 const path_len: u32 = @intCast(remote_path.len);
703 try wire_buf.appendSlice(alloc, std.mem.asBytes(&path_len));
704 try wire_buf.appendSlice(alloc, remote_path);
705 try wire_buf.appendSlice(alloc, content);
706
707 ipc.send(result.fd, .Write, wire_buf.items) catch |err| switch (err) {
708 error.BrokenPipe, error.ConnectionResetByPeer => return,
709 else => return err,
710 };
711
712 var sb = try ipc.SocketBuffer.init(alloc);
713 defer sb.deinit();
714
715 const n = sb.read(result.fd) catch return error.ReadFailed;
716 if (n == 0) return error.ConnectionClosed;
717
718 while (sb.next()) |msg| {
719 if (msg.header.tag == .Ack) return;
720 }
721 return error.NoAckReceived;
722}
723
724/// Fetch session scrollback with optional row count.
725pub fn historyRange(alloc: Allocator, io: Io, name: []const u8, format: HistoryFormat, scrollback_rows: ?u32) ![]u8 {
726 signal.ignoreSigpipe();
727 const sesh = try socket.getSeshName(alloc, name);
728 defer alloc.free(sesh);
729
730 var cfg = try Cfg.init(alloc, io);
731 defer cfg.deinit(alloc);
732
733 const socket_path = try socket.getSocketPath(alloc, cfg.socket_dir, sesh);
734 defer alloc.free(socket_path);
735
736 var dir = try std.Io.Dir.openDirAbsolute(io, cfg.socket_dir, .{});
737 defer dir.close(io);
738
739 const exists = try socket.sessionExists(io, dir, sesh);
740 if (!exists) return error.SessionNotFound;
741
742 const fd = ipc.connectSession(socket_path) catch |err| {
743 if (err == error.ConnectionRefused) socket.cleanupStaleSocket(io, dir, sesh);
744 return error.SessionUnresponsive;
745 };
746 defer lib_posix.close(fd);
747
748 const reply_tag: ipc.Tag = if (scrollback_rows != null) .Capture else .History;
749 if (scrollback_rows) |rows| {
750 var request = std.mem.zeroes(ipc.Capture);
751 request.format = @backingInt(format);
752 request.rows = rows;
753 try ipc.send(fd, .Capture, std.mem.asBytes(&request));
754 try ipc.send(fd, .Info, "");
755 } else {
756 const format_byte = [_]u8{@backingInt(format)};
757 try ipc.send(fd, .History, &format_byte);
758 }
759
760 var sb = try ipc.SocketBuffer.init(alloc);
761 defer sb.deinit();
762
763 while (true) {
764 var poll_fds = [_]lib_posix.pollfd{.{ .fd = fd, .events = lib_posix.POLL.IN, .revents = 0 }};
765 const poll_result = lib_posix.poll(&poll_fds, 5000) catch return error.Timeout;
766 if (poll_result == 0) return error.Timeout;
767
768 const n = sb.read(fd) catch return error.ReadFailed;
769 if (n == 0) return error.ConnectionClosed;
770
771 while (sb.next()) |msg| {
772 if (msg.header.tag == reply_tag) {
773 return alloc.dupe(u8, msg.payload);
774 }
775 if (scrollback_rows != null and msg.header.tag == .Info) {
776 return error.CaptureNotSupported;
777 }
778 }
779 }
780}
781
782/// Fetch session scrollback in `.plain`, `.vt`, or `.html` format. Caller owns slice.
783pub fn history(alloc: Allocator, io: Io, name: []const u8, format: HistoryFormat) ![]u8 {
784 return historyRange(alloc, io, name, format, null);
785}
786
787/// Get session labels (all `k=v` pairs if `key == null`, or single value). Caller owns slice.
788pub fn get(alloc: Allocator, io: Io, name: []const u8, key: ?[]const u8) ![]u8 {
789 signal.ignoreSigpipe();
790 const sesh = try socket.getSeshName(alloc, name);
791 defer alloc.free(sesh);
792
793 var cfg = try Cfg.init(alloc, io);
794 defer cfg.deinit(alloc);
795
796 const socket_path = try socket.getSocketPath(alloc, cfg.socket_dir, sesh);
797 defer alloc.free(socket_path);
798
799 const payload = try ipc.roundTripForTag(alloc, socket_path, .LabelGet, "", .LabelData);
800 errdefer alloc.free(payload);
801
802 if (key == null) return payload;
803
804 const val = try label.getLabelValueFromPairs(key.?, payload);
805 const result = try alloc.dupe(u8, val);
806 alloc.free(payload);
807 return result;
808}
809
810/// Set or unset (`k=`) space-separated session labels.
811pub fn set(alloc: Allocator, io: Io, name: []const u8, kvs: []const u8) !void {
812 signal.ignoreSigpipe();
813 var it = label.LabelIterator.init(kvs);
814 while (it.next()) |kv| {
815 try label.assertLabel(kv.key, kv.value);
816 }
817
818 const sesh = try socket.getSeshName(alloc, name);
819 defer alloc.free(sesh);
820
821 var cfg = try Cfg.init(alloc, io);
822 defer cfg.deinit(alloc);
823
824 const socket_path = try socket.getSocketPath(alloc, cfg.socket_dir, sesh);
825 defer alloc.free(socket_path);
826
827 _ = try ipc.roundTripForTag(alloc, socket_path, .LabelSet, kvs, .Ack);
828}
829
830/// Clear all labels on a session.
831pub fn clear(alloc: Allocator, io: Io, name: []const u8) !void {
832 signal.ignoreSigpipe();
833 const sesh = try socket.getSeshName(alloc, name);
834 defer alloc.free(sesh);
835
836 var cfg = try Cfg.init(alloc, io);
837 defer cfg.deinit(alloc);
838
839 const socket_path = try socket.getSocketPath(alloc, cfg.socket_dir, sesh);
840 defer alloc.free(socket_path);
841
842 _ = try ipc.roundTripForTag(alloc, socket_path, .LabelClear, "", .Ack);
843}
844
845/// Get tracked environment variables for the session leader (`zmx print-env`).
846pub fn printEnv(alloc: Allocator, io: Io, name: []const u8, key: ?[]const u8) ![]u8 {
847 signal.ignoreSigpipe();
848 const sesh = try socket.getSeshName(alloc, name);
849 defer alloc.free(sesh);
850
851 var cfg = try Cfg.init(alloc, io);
852 defer cfg.deinit(alloc);
853
854 const socket_path = try socket.getSocketPath(alloc, cfg.socket_dir, sesh);
855 defer alloc.free(socket_path);
856
857 const payload = try ipc.roundTripForTag(alloc, socket_path, .EnvGet, "", .EnvData);
858
859 // A whole-environment query gives this buffer to the caller.
860 if (key == null) return payload;
861 defer alloc.free(payload);
862
863 // Search for key in payload: lines formatted as "KEY=VALUE\n"
864 var lines = std.mem.splitScalar(u8, payload, '\n');
865 while (lines.next()) |line| {
866 if (std.mem.startsWith(u8, line, key.?)) {
867 if (line.len > key.?.len and line[key.?.len] == '=') {
868 const val = line[key.?.len + 1 ..];
869 return try alloc.dupe(u8, val);
870 }
871 }
872 }
873 return error.EnvVarNotFound;
874}