Eric Bower
·
2026-08-11
1const std = @import("std");
2const build_options = @import("build_options");
3const ghostty_vt = @import("ghostty-vt");
4const ipc = @import("ipc.zig");
5const log = @import("log.zig");
6const completions = @import("completions.zig");
7const util = @import("util.zig");
8const cross = @import("cross.zig");
9const socket = @import("socket.zig");
10const label = @import("label.zig");
11const lib_posix = @import("posix.zig");
12const signal = @import("signal.zig");
13const Cfg = @import("cfg.zig");
14const loop = @import("loop.zig");
15const Client = loop.Client;
16const Daemon = loop.Daemon;
17const version = build_options.version;
18const ghostty_version = build_options.ghostty_version;
19
20pub const std_options: std.Options = .{
21 .logFn = log.zmxLogFn,
22 .log_level = .debug,
23};
24
25/// This is the entry point for the CLI.
26pub fn main(init: std.process.Init) !void {
27 const gpa = init.gpa;
28 const io = init.io;
29
30 // Every subcommand may write to a Unix-domain socket; a peer that
31 // disappears between probe and send would otherwise kill us before
32 // write() can return BrokenPipe. Inherited across fork, so this also
33 // covers the daemon.
34 signal.ignoreSigpipe();
35
36 var args = init.minimal.args.iterate();
37 defer args.deinit();
38 _ = args.next(); // skip program name
39
40 var cfg = try Cfg.init(gpa, io);
41 defer cfg.deinit(gpa);
42
43 const log_path = try std.fs.path.join(gpa, &.{ cfg.log_dir, "zmx.log" });
44 defer gpa.free(log_path);
45 const log_mode = std.Io.File.Permissions.fromMode(@intCast(cfg.log_mode));
46 try log.log_system.init(io, log_path, log_mode);
47 defer log.log_system.deinit();
48
49 const shell_env = init.environ_map.get("SHELL") orelse "/bin/sh";
50
51 const cmd = args.next() orelse {
52 return list(gpa, io, &cfg, false);
53 };
54
55 if (std.mem.eql(u8, cmd, "version") or std.mem.eql(u8, cmd, "v") or std.mem.eql(u8, cmd, "-v") or std.mem.eql(u8, cmd, "--version")) {
56 return printVersion(io, &cfg);
57 } else if (std.mem.eql(u8, cmd, "help") or std.mem.eql(u8, cmd, "h") or std.mem.eql(u8, cmd, "-h")) {
58 return help(io);
59 } else if (std.mem.eql(u8, cmd, "list") or std.mem.eql(u8, cmd, "l") or std.mem.eql(u8, cmd, "ls")) {
60 var short = false;
61 while (args.next()) |arg| {
62 if (detectHelp(arg)) return help(io);
63 if (std.mem.eql(u8, arg, "--short")) short = true;
64 }
65 return list(gpa, io, &cfg, short);
66 } else if (std.mem.eql(u8, cmd, "get") or std.mem.eql(u8, cmd, "g")) {
67 const sesh_name = args.next() orelse return error.SessionNameRequired;
68 if (detectHelp(sesh_name)) return help(io);
69 const sesh = try socket.resolveSessionOrEnv(gpa, io, sesh_name);
70 defer gpa.free(sesh);
71 const single_kv = args.next() orelse "";
72 return labelGet(gpa, io, &cfg, sesh, single_kv);
73 } else if (std.mem.eql(u8, cmd, "set")) {
74 const sesh_name = args.next() orelse return error.SessionNameRequired;
75 if (detectHelp(sesh_name)) return help(io);
76 const sesh = try socket.resolveSessionOrEnv(gpa, io, sesh_name);
77 defer gpa.free(sesh);
78
79 var kvs = std.ArrayList(u8).empty;
80 defer kvs.deinit(gpa);
81 var first = true;
82 while (args.next()) |arg| {
83 if (!first) try kvs.append(gpa, ' ');
84 try kvs.appendSlice(gpa, arg);
85 first = false;
86 }
87 return labelSet(gpa, io, &cfg, sesh, kvs.items);
88 } else if (std.mem.eql(u8, cmd, "clear")) {
89 const sesh_name = args.next() orelse return error.SessionNameRequired;
90 if (detectHelp(sesh_name)) return help(io);
91 const sesh = try socket.resolveSessionOrEnv(gpa, io, sesh_name);
92 defer gpa.free(sesh);
93 return labelClear(gpa, io, &cfg, sesh);
94 } else if (std.mem.eql(u8, cmd, "completions") or std.mem.eql(u8, cmd, "c")) {
95 const arg = args.next() orelse return;
96 if (std.mem.eql(u8, arg, "--help") or std.mem.eql(u8, arg, "-h")) {
97 return help(io);
98 }
99 const shell = completions.Shell.fromString(arg) orelse return;
100 return printCompletions(io, shell);
101 } else if (std.mem.eql(u8, cmd, "detach") or std.mem.eql(u8, cmd, "d")) {
102 return detachAll(gpa, io, &cfg);
103 } else if (std.mem.eql(u8, cmd, "history") or std.mem.eql(u8, cmd, "hi")) {
104 var session_name: ?[]const u8 = null;
105 var format: util.HistoryFormat = .plain;
106 while (args.next()) |arg| {
107 if (std.mem.eql(u8, arg, "--help") or std.mem.eql(u8, arg, "-h")) {
108 return help(io);
109 } else if (std.mem.eql(u8, arg, "--vt")) {
110 format = .vt;
111 } else if (std.mem.eql(u8, arg, "--html")) {
112 format = .html;
113 } else if (session_name == null) {
114 session_name = arg;
115 }
116 }
117 const sesh_env = socket.getSeshNameFromEnv();
118 const sesh = try socket.getSeshName(gpa, session_name orelse sesh_env);
119 defer gpa.free(sesh);
120 return history(gpa, io, &cfg, sesh, format);
121 } else if (std.mem.eql(u8, cmd, "attach") or std.mem.eql(u8, cmd, "a")) {
122 const session_name = args.next() orelse "";
123 if (std.mem.eql(u8, session_name, "--help") or std.mem.eql(u8, session_name, "-h")) {
124 return help(io);
125 }
126
127 var command_args: std.ArrayList([]const u8) = .empty;
128 defer command_args.deinit(gpa);
129 while (args.next()) |arg| {
130 try command_args.append(gpa, arg);
131 }
132
133 var command: ?[][]const u8 = null;
134 if (command_args.items.len > 0) {
135 command = command_args.items;
136 }
137
138 var cwd_buf: [std.fs.max_path_bytes]u8 = undefined;
139 const cwd_len = std.process.currentPath(io, &cwd_buf) catch 0;
140 const cwd = cwd_buf[0..cwd_len];
141
142 const sesh = try socket.getSeshName(gpa, session_name);
143 defer gpa.free(sesh);
144 const socket_path = socket.getSocketPath(gpa, cfg.socket_dir, sesh) catch |err| switch (err) {
145 error.NameTooLong => return socket.printSessionNameTooLong(io, sesh, cfg.socket_dir),
146 error.OutOfMemory => return err,
147 };
148 var daemon = Daemon.init(io, &cfg, sesh, socket_path);
149 daemon.command = command;
150 daemon.setCwd(cwd);
151 daemon.shell = shell_env;
152 std.log.info("socket path={s}", .{daemon.socket_path});
153 return attach(gpa, io, &daemon);
154 } else if (std.mem.eql(u8, cmd, "run") or std.mem.eql(u8, cmd, "r")) {
155 const session_name = args.next() orelse "";
156 if (std.mem.eql(u8, session_name, "--help") or std.mem.eql(u8, session_name, "-h")) {
157 return help(io);
158 }
159
160 var cmd_args_raw: std.ArrayList([]const u8) = .empty;
161 defer cmd_args_raw.deinit(gpa);
162 var detached = false;
163 while (args.next()) |arg| {
164 if (std.mem.startsWith(u8, arg, "-d")) {
165 detached = true;
166 } else {
167 try cmd_args_raw.append(gpa, arg);
168 }
169 }
170
171 var cwd_buf: [std.fs.max_path_bytes]u8 = undefined;
172 const cwd_len = std.process.currentPath(io, &cwd_buf) catch 0;
173 const cwd = cwd_buf[0..cwd_len];
174
175 const sesh = try socket.getSeshName(gpa, session_name);
176 defer gpa.free(sesh);
177 const socket_path = socket.getSocketPath(gpa, cfg.socket_dir, sesh) catch |err| switch (err) {
178 error.NameTooLong => return socket.printSessionNameTooLong(io, sesh, cfg.socket_dir),
179 error.OutOfMemory => return err,
180 };
181 defer gpa.free(socket_path);
182 var daemon = Daemon.init(io, &cfg, sesh, socket_path);
183 daemon.setCwd(cwd);
184 daemon.is_task_mode = true;
185 daemon.shell = shell_env;
186 std.log.info("socket path={s}", .{daemon.socket_path});
187 return run(gpa, io, &daemon, detached, cmd_args_raw.items);
188 } else if (std.mem.eql(u8, cmd, "send") or std.mem.eql(u8, cmd, "s")) {
189 const session_name = args.next() orelse "";
190 if (std.mem.eql(u8, session_name, "--help") or std.mem.eql(u8, session_name, "-h")) {
191 return help(io);
192 }
193 if (session_name.len == 0) return error.SessionNameRequired;
194
195 var text_parts: std.ArrayList([]const u8) = .empty;
196 defer text_parts.deinit(gpa);
197 while (args.next()) |arg| {
198 try text_parts.append(gpa, arg);
199 }
200
201 const sesh = try socket.getSeshName(gpa, session_name);
202 defer gpa.free(sesh);
203 const socket_path = socket.getSocketPath(gpa, cfg.socket_dir, sesh) catch |err| switch (err) {
204 error.NameTooLong => return socket.printSessionNameTooLong(io, sesh, cfg.socket_dir),
205 error.OutOfMemory => return err,
206 };
207 return send(gpa, io, &cfg, sesh, socket_path, text_parts.items, .Send);
208 } else if (std.mem.eql(u8, cmd, "print") or std.mem.eql(u8, cmd, "p")) {
209 const session_name = args.next() orelse "";
210 if (std.mem.eql(u8, session_name, "--help") or std.mem.eql(u8, session_name, "-h")) {
211 return help(io);
212 }
213 if (session_name.len == 0) return error.SessionNameRequired;
214
215 var text_parts: std.ArrayList([]const u8) = .empty;
216 defer text_parts.deinit(gpa);
217 while (args.next()) |arg| {
218 try text_parts.append(gpa, arg);
219 }
220
221 const sesh = try socket.getSeshName(gpa, session_name);
222 defer gpa.free(sesh);
223 const socket_path = socket.getSocketPath(gpa, cfg.socket_dir, sesh) catch |err| switch (err) {
224 error.NameTooLong => return socket.printSessionNameTooLong(io, sesh, cfg.socket_dir),
225 error.OutOfMemory => return err,
226 };
227 return send(gpa, io, &cfg, sesh, socket_path, text_parts.items, .Output);
228 } else if (std.mem.eql(u8, cmd, "kill") or std.mem.eql(u8, cmd, "k")) {
229 var stderr_buffer: [1024]u8 = undefined;
230 var stderr_writer = std.Io.File.stderr().writer(io, &stderr_buffer);
231 const stderr = &stderr_writer.interface;
232
233 var matchers: std.ArrayList(socket.SessionMatch) = .empty;
234 defer {
235 for (matchers.items) |m| {
236 gpa.free(m.name);
237 }
238 matchers.deinit(gpa);
239 }
240 var force = false;
241 while (args.next()) |session_name| {
242 if (std.mem.eql(u8, session_name, "--help") or std.mem.eql(u8, session_name, "-h")) {
243 return help(io);
244 }
245 if (std.mem.eql(u8, session_name, "--force")) {
246 force = true;
247 continue;
248 }
249 const m = try socket.parseSessionArg(gpa, session_name);
250 try matchers.append(gpa, m);
251 }
252 if (matchers.items.len == 0) {
253 return error.SessionNameRequired;
254 }
255 var sessions = try util.get_session_entries(gpa, io, cfg.socket_dir);
256 defer {
257 for (sessions.items) |session| {
258 session.deinit(gpa);
259 }
260 sessions.deinit(gpa);
261 }
262
263 for (sessions.items) |session| {
264 for (matchers.items) |m| {
265 if (!m.matches(session.name)) {
266 continue;
267 }
268
269 kill(gpa, io, &cfg, session.name, force) catch |err| {
270 try stderr.print(
271 "failed to kill session={s}: {s}\n",
272 .{ session.name, @errorName(err) },
273 );
274 try stderr.flush();
275 };
276 break;
277 }
278 }
279 } else if (std.mem.eql(u8, cmd, "wait") or std.mem.eql(u8, cmd, "w")) {
280 var matchers: std.ArrayList(socket.SessionMatch) = .empty;
281 defer {
282 for (matchers.items) |m| {
283 gpa.free(m.name);
284 }
285 matchers.deinit(gpa);
286 }
287 while (args.next()) |session_name| {
288 if (std.mem.eql(u8, session_name, "--help") or std.mem.eql(u8, session_name, "-h")) {
289 return help(io);
290 }
291 const m = try socket.parseSessionArg(gpa, session_name);
292 try matchers.append(gpa, m);
293 }
294 if (matchers.items.len == 0) {
295 return error.SessionNameRequired;
296 }
297 return wait(gpa, io, &cfg, matchers);
298 } else if (std.mem.eql(u8, cmd, "tail") or std.mem.eql(u8, cmd, "t")) {
299 var matchers: std.ArrayList(socket.SessionMatch) = .empty;
300 defer {
301 for (matchers.items) |m| {
302 gpa.free(m.name);
303 }
304 matchers.deinit(gpa);
305 }
306 while (args.next()) |session_name| {
307 if (std.mem.eql(u8, session_name, "--help") or std.mem.eql(u8, session_name, "-h")) {
308 return help(io);
309 }
310 const m = try socket.parseSessionArg(gpa, session_name);
311 try matchers.append(gpa, m);
312 }
313 if (matchers.items.len == 0) {
314 return error.SessionNameRequired;
315 }
316
317 // Resolve matchers against session list to get actual session names.
318 var resolved_names: std.ArrayList([]const u8) = .empty;
319 defer {
320 for (resolved_names.items) |name| {
321 gpa.free(name);
322 }
323 resolved_names.deinit(gpa);
324 }
325
326 var any_prefix = false;
327 for (matchers.items) |m| {
328 if (m.is_prefix) {
329 any_prefix = true;
330 break;
331 }
332 }
333
334 if (any_prefix) {
335 var sessions = try util.get_session_entries(gpa, io, cfg.socket_dir);
336 defer {
337 for (sessions.items) |session| {
338 session.deinit(gpa);
339 }
340 sessions.deinit(gpa);
341 }
342 for (sessions.items) |session| {
343 for (matchers.items) |m| {
344 if (m.matches(session.name)) {
345 try resolved_names.append(gpa, try gpa.dupe(u8, session.name));
346 break;
347 }
348 }
349 }
350 }
351 // Add exact-match names directly.
352 for (matchers.items) |m| {
353 if (!m.is_prefix) {
354 try resolved_names.append(gpa, try gpa.dupe(u8, m.name));
355 }
356 }
357
358 var client_socket_fds = try std.ArrayList(i32).initCapacity(gpa, resolved_names.items.len);
359 defer {
360 for (client_socket_fds.items) |client_fd| {
361 lib_posix.close(client_fd);
362 }
363 client_socket_fds.deinit(gpa);
364 }
365
366 for (resolved_names.items) |session_name| {
367 const socket_path = socket.getSocketPath(gpa, cfg.socket_dir, session_name) catch |err| switch (err) {
368 error.NameTooLong => return socket.printSessionNameTooLong(init.io, session_name, cfg.socket_dir),
369 error.OutOfMemory => return err,
370 };
371 const client_sock = try socket.sessionConnect(socket_path);
372 try client_socket_fds.append(gpa, client_sock);
373 }
374 _ = try tail(gpa, client_socket_fds, false, false);
375 } else if (std.mem.eql(u8, cmd, "write") or std.mem.eql(u8, cmd, "wr")) {
376 const session_name = args.next() orelse "";
377 if (std.mem.eql(u8, session_name, "--help") or std.mem.eql(u8, session_name, "-h")) {
378 return help(io);
379 }
380 if (session_name.len == 0) return error.SessionNameRequired;
381 const file_path = args.next() orelse "";
382 if (std.mem.eql(u8, file_path, "--help") or std.mem.eql(u8, file_path, "-h")) {
383 return help(io);
384 }
385 if (file_path.len == 0) return error.FilePathRequired;
386
387 var cwd_buf: [std.fs.max_path_bytes]u8 = undefined;
388 const cwd_len = std.process.currentPath(io, &cwd_buf) catch 0;
389 const cwd = cwd_buf[0..cwd_len];
390 const sesh = try socket.getSeshName(gpa, session_name);
391 defer gpa.free(sesh);
392 const socket_path = socket.getSocketPath(gpa, cfg.socket_dir, sesh) catch |err| switch (err) {
393 error.NameTooLong => return socket.printSessionNameTooLong(io, sesh, cfg.socket_dir),
394 error.OutOfMemory => return err,
395 };
396 var daemon = Daemon.init(io, &cfg, sesh, socket_path);
397 daemon.is_task_mode = true;
398 daemon.setCwd(cwd);
399 daemon.shell = shell_env;
400 std.log.info("socket path={s}", .{daemon.socket_path});
401 try writeFile(gpa, io, &daemon, file_path);
402 } else {
403 return help(io);
404 }
405}
406
407fn help(io: std.Io) !void {
408 const help_text =
409 \\zmx - session persistence for terminal processes
410 \\
411 \\Usage: zmx <command> [args...]
412 \\
413 \\Commands:
414 \\ [a]ttach <name> [command...] Attach to session, creating if needed
415 \\ [r]un <name> [-d] [command...] Send command without attaching
416 \\ [s]end <name> <text...> Send raw input to session PTY
417 \\ [p]rint <name> <text...> Inject text into session display
418 \\ [wr]ite <name> <file_path> Write stdin to file_path through the session
419 \\ [d]etach Detach all clients (ctrl+\\ for current client)
420 \\ [l]ist|ls [--short|--where k=v] List active sessions
421 \\ [g]et <name> Get session labels
422 \\ set <name> k=v ... Set session labels (k= to remove)
423 \\ [cl]ear <name> Clear all session labels
424 \\ [k]ill <name>... [--force] Kill session and all attached clients
425 \\ [hi]story <name> [--vt|--html] Output session scrollback
426 \\ [w]ait <name>... Wait for session tasks to complete
427 \\ [t]ail <name>... Follow session output
428 \\ [c]ompletions <shell> Shell completions (bash, zsh, fish, nu)
429 \\ [v]ersion Show version and metadata (socket dir, log dir)
430 \\ [h]elp Show this help
431 \\
432 \\Attach:
433 \\ This will spawn a login $SHELL with a PTY. You can provide a
434 \\ command instead of creating a shell.
435 \\
436 \\ Examples:
437 \\ zmx attach dev
438 \\ zmx attach dev vim
439 \\
440 \\History:
441 \\ This should generally be used with `tail` to print the last lines
442 \\ of the session's scrollback history.
443 \\
444 \\ Examples:
445 \\ zmx history <session> | tail -100
446 \\
447 \\Run:
448 \\ Commands run inside a PTY using bash
449 \\ Commands are passed as-is: do not wrap in quotes.
450 \\ Commands run sequentially: do not send multiple in parallel.
451 \\ Stdin is redirected from /dev/null to prevent interactive programs
452 \\ (pagers, editors, prompts) from blocking. Use `zmx send` for
453 \\ commands that need user input, or pipe data directly:
454 \\ echo "data" | zmx run dev cat
455 \\
456 \\ `-d` will detach from the calling terminal. Use `wait` to track
457 \\ its status.
458 \\
459 \\ Examples:
460 \\ zmx run dev ls
461 \\ zmx run dev zig build
462 \\ zmx run dev grep -r TODO src
463 \\ zmx run dev git log --oneline # pager won't block
464 \\ echo "hello" | zmx run dev cat # piped stdin still works
465 \\
466 \\ # heredoc
467 \\ printf "cat << 'EOF'\r\nHello $USER\r\nToday is $(date).\r\nEOF" | zmx run dev
468 \\
469 \\ # non-blocking
470 \\ zmx run dev -d sleep 10
471 \\ zmx wait dev
472 \\
473 \\Send:
474 \\ Sends raw text to the session's PTY input (fire-and-forget).
475 \\ Unlike `run`, no completion marker is appended and no exit code
476 \\ is tracked. Useful for TUI applications, interactive prompts,
477 \\ or any program that reads stdin directly.
478 \\
479 \\ Text is sent byte-for-byte with no automatic carriage return.
480 \\ Append \r yourself when you want the shell to execute a command.
481 \\
482 \\ Text can also be piped via stdin:
483 \\ printf 'ls -la\r' | zmx send dev
484 \\
485 \\ Examples:
486 \\ printf 'echo hello\r' | zmx send dev
487 \\ zmx send dev $(printf '\x03')
488 \\ zmx send dev /compact
489 \\
490 \\Print:
491 \\ Injects text directly into the session display and scrollback.
492 \\ Never touches the PTY input -- the shell sees nothing.
493 \\ Caller is responsible for newlines (\\r\\n).
494 \\
495 \\ Examples:
496 \\ printf '\\r\\nhello\\r\\n' | zmx print dev
497 \\ zmx print dev "$(printf '\\r\\nalert\\r\\n')"
498 \\
499 \\Write:
500 \\ Writes stdin to file_path inside the session. Works over SSH.
501 \\ file_path can be absolute or relative to the session shell's cwd.
502 \\ Requires base64 and printf in the remote environment.
503 \\ Large files are chunked automatically (~48KB per chunk).
504 \\ File path must not contain single quotes.
505 \\
506 \\ Examples:
507 \\ echo "hello" | zmx write dev /tmp/hello.txt
508 \\ cat main.zig | zmx write dev src/main.zig
509 \\
510 \\Wait:
511 \\ Used with a detached run task to track its status. Multiple
512 \\ sessions can be provided.
513 \\
514 \\ Examples:
515 \\ zmx run -d dev sleep 10
516 \\ zmx wait dev
517 \\ zmx wait dev other
518 \\
519 \\Labels:
520 \\ Attach key=value labels to live sessions for discovery and
521 \\ filtering. Labels are in-memory and scoped to session lifetime.
522 \\
523 \\ Examples:
524 \\ zmx set dev project=zmx env=dev
525 \\ zmx set dev project= # unset a label
526 \\ zmx set . status=fail # "." resolves to current session
527 \\ zmx get dev
528 \\ zmx get dev project
529 \\ zmx set next "$(zmx get prev)" # set labels from other session
530 \\ zmx list | grep project=zmx
531 \\ zmx clear dev
532 \\
533 \\Environment variables:
534 \\ SHELL Default shell for new sessions
535 \\ ZMX_DIR Socket directory (priority 1)
536 \\ XDG_RUNTIME_DIR Socket directory (priority 2)
537 \\ TMPDIR Socket directory (priority 3)
538 \\ ZMX_SESSION Session name (injected automatically)
539 \\ ZMX_SESSION_PREFIX Prefix added to all session names
540 \\ ZMX_DIR_MODE Sets mode for socket and log directories (octal, defaults to 0750)
541 \\ ZMX_LOG_MODE Sets mode for log files (octal, defaults to 0640)
542 \\ ZMX_NO_DETACH_KEY Disables the ctrl+\ detach shortcut (set to any value)
543 \\
544 ;
545 var buf: [8192]u8 = undefined;
546 var w = std.Io.File.stdout().writer(io, &buf);
547 try w.interface.print(help_text, .{});
548 try w.interface.flush();
549}
550
551fn printVersion(io: std.Io, cfg: *Cfg) !void {
552 var buf: [256]u8 = undefined;
553 var w = std.Io.File.stdout().writer(io, &buf);
554 try w.interface.print(
555 "zmx\t\t{s}\nghostty_vt\t{s}\nsocket_dir\t{s}\nlog_dir\t\t{s}\n",
556 .{ version, ghostty_version, cfg.socket_dir, cfg.log_dir },
557 );
558 try w.interface.flush();
559}
560
561fn printCompletions(io: std.Io, shell: completions.Shell) !void {
562 const script = shell.getCompletionScript();
563 var buf: [8192]u8 = undefined;
564 var w = std.Io.File.stdout().writer(io, &buf);
565 try w.interface.print("{s}\n", .{script});
566 try w.interface.flush();
567}
568
569fn detectHelp(arg: []const u8) bool {
570 return (std.mem.eql(u8, arg, "--help") or std.mem.eql(u8, arg, "-h"));
571}
572
573fn tail(alloc: std.mem.Allocator, client_socket_fds: std.ArrayList(i32), detached: bool, is_run_cmd: bool) !u8 {
574 var poll_fds = try std.ArrayList(lib_posix.pollfd).initCapacity(alloc, 4);
575 defer poll_fds.deinit(alloc);
576
577 var read_buf = try ipc.SocketBuffer.init(alloc);
578 defer read_buf.deinit();
579
580 var stdout_buf = try std.ArrayList(u8).initCapacity(alloc, 4096);
581 defer stdout_buf.deinit(alloc);
582
583 var is_first_line = true;
584 var task_complete_code: ?u8 = null;
585
586 while (true) {
587 poll_fds.clearRetainingCapacity();
588
589 // Poll socket for read
590 for (client_socket_fds.items) |client_sock_fd| {
591 try poll_fds.append(alloc, .{
592 .fd = client_sock_fd,
593 .events = lib_posix.POLL.IN,
594 .revents = 0,
595 });
596 }
597
598 // Poll for write if we have pending data
599 if (stdout_buf.items.len > 0) {
600 try poll_fds.append(alloc, .{
601 .fd = lib_posix.STDOUT_FILENO,
602 .events = lib_posix.POLL.OUT,
603 .revents = 0,
604 });
605 }
606
607 _ = lib_posix.poll(poll_fds.items, -1) catch |err| {
608 if (err == error.Interrupted) continue; // EINTR from signal, loop again
609 return err;
610 };
611
612 // Handle socket read (incoming Output messages from daemon)
613 for (poll_fds.items) |*poll_fd| {
614 if (poll_fd.revents & lib_posix.POLL.IN != 0) {
615 const n = read_buf.read(poll_fd.fd) catch |err| {
616 if (err == error.WouldBlock) continue;
617 if (err == error.ConnectionResetByPeer or err == error.BrokenPipe) {
618 return 1;
619 }
620 std.log.err("daemon read err={s}", .{@errorName(err)});
621 return err;
622 };
623 if (n == 0) {
624 // Server closed connection. If we got task completion,
625 // return the exit code. Otherwise fall back to 0.
626 if (task_complete_code) |exit_code| {
627 return exit_code;
628 }
629 return 0;
630 }
631
632 while (read_buf.next()) |msg| {
633 switch (msg.header.tag) {
634 .Ack => {
635 if (detached) {
636 _ = lib_posix.write(lib_posix.STDOUT_FILENO, "command sent!\n") catch |err| blk: {
637 if (err == error.WouldBlock) break :blk 0;
638 return err;
639 };
640 return 0;
641 }
642 },
643 .Output => {
644 if (msg.payload.len > 0) {
645 // TODO: figure out how to bring this back
646 // Fallback: scan output for task exit marker in case
647 // .TaskComplete was lost (e.g. daemon exited before
648 // flushing). This ensures we detect completion even
649 // when the IPC message doesn't arrive.
650 // if (task_complete_code == null and is_run_cmd) {
651 // if (util.findTaskExitMarker(msg.payload)) |ec| {
652 // task_complete_code = ec;
653 // }
654 // }
655
656 // Strip the first line (command echo) for run mode.
657 var payload = msg.payload;
658 if (!detached and is_run_cmd and is_first_line) {
659 if (std.mem.indexOfScalar(u8, payload, '\n')) |nl| {
660 is_first_line = false;
661 payload = payload[nl + 1 ..];
662 } else {
663 is_first_line = false;
664 payload = payload[payload.len..]; // consume entire echo line
665 }
666 }
667
668 if (payload.len > 0) {
669 // Strip ANSI escape sequences to produce plain text.
670 // This prevents shell prompts, colors, cursor movements,
671 // and other VT sequences from corrupting the caller's terminal.
672 const plain = util.stripAnsi(alloc, payload) catch |err| {
673 std.log.warn("stripAnsi failed: {s}", .{@errorName(err)});
674 continue;
675 };
676 defer alloc.free(plain);
677 if (plain.len > 0) {
678 try stdout_buf.appendSlice(alloc, plain);
679 }
680 }
681 }
682 },
683 .TaskComplete => {
684 task_complete_code = if (msg.payload.len > 0) msg.payload[0] else 0;
685 },
686 else => {},
687 }
688 }
689 }
690 }
691
692 // Check for task completion after processing socket messages.
693 // This must be outside the stdout write block because .TaskComplete
694 // can arrive after all output has already been flushed, leaving
695 // stdout_buf empty. Without this check, tail() would poll forever.
696 if (task_complete_code) |exit_code| {
697 // Flush any remaining output before returning
698 flush_loop: while (stdout_buf.items.len > 0) {
699 const n = lib_posix.write(lib_posix.STDOUT_FILENO, stdout_buf.items) catch |err| {
700 if (err == error.WouldBlock) break :flush_loop;
701 return err;
702 };
703 try stdout_buf.replaceRange(alloc, 0, n, &[_]u8{});
704 }
705 return exit_code;
706 }
707
708 if (stdout_buf.items.len > 0) {
709 const n = lib_posix.write(lib_posix.STDOUT_FILENO, stdout_buf.items) catch |err| blk: {
710 if (err == error.WouldBlock) break :blk 0;
711 return err;
712 };
713 if (n > 0) {
714 try stdout_buf.replaceRange(alloc, 0, n, &[_]u8{});
715 }
716 }
717
718 // Check for HUP/ERR on any socket
719 for (poll_fds.items) |poll_fd| {
720 if (poll_fd.revents & (lib_posix.POLL.HUP | lib_posix.POLL.ERR | lib_posix.POLL.NVAL) != 0) {
721 return 0;
722 }
723 }
724 }
725}
726
727fn wait(alloc: std.mem.Allocator, io: std.Io, cfg: *Cfg, matchers: std.ArrayList(socket.SessionMatch)) !void {
728 var stdout_buffer: [1024]u8 = undefined;
729 var stdout_writer = std.Io.File.stdout().writer(io, &stdout_buffer);
730 const stdout = &stdout_writer.interface;
731
732 var stderr_buffer: [1024]u8 = undefined;
733 var stderr_writer = std.Io.File.stderr().writer(io, &stderr_buffer);
734 const stderr = &stderr_writer.interface;
735
736 // Highest match count seen so far. Lets us distinguish "sessions haven't
737 // appeared yet" (keep polling) from "sessions we were tracking
738 // disappeared" (fail -- daemon crashed or was killed).
739 var max_seen: i32 = 0;
740 var zero_match_iters: u32 = 0;
741
742 var agg_exit_code: u8 = 0;
743 var last_print: std.Io.Timestamp = .zero;
744 var prev_done: i32 = 0;
745 while (true) {
746 agg_exit_code = 0;
747 var sessions = try util.get_session_entries(alloc, io, cfg.socket_dir);
748 var total: i32 = 0;
749 var done: i32 = 0;
750
751 for (sessions.items) |session| {
752 var found = false;
753 for (matchers.items) |m| {
754 if (m.matches(session.name)) {
755 found = true;
756 break;
757 }
758 }
759 if (!found) {
760 continue;
761 }
762
763 total += 1;
764 if (session.is_error) {
765 // Daemon unreachable (probe timed out). On Timeout the socket
766 // is no longer deleted, so this session would otherwise
767 // persist as task_ended_at==0 forever → infinite "still
768 // waiting". Count it as done+failed so wait terminates.
769 try stderr.print(
770 "[{d}] task unreachable: {s} ({s})\n",
771 .{
772 std.Io.Timestamp.now(io, .real).toSeconds(),
773 session.name,
774 session.error_name orelse "unknown",
775 },
776 );
777 try stderr.flush();
778 agg_exit_code = 1;
779 done += 1;
780 continue;
781 }
782 if (session.task_ended_at == 0) {
783 const now = std.Io.Timestamp.now(io, .real);
784 if (now.toSeconds() - last_print.toSeconds() >= 5) {
785 try stdout.print(
786 "[{d}] waiting task={s}\n",
787 .{ now.toSeconds(), session.name },
788 );
789 try stdout.flush();
790 last_print = now;
791 }
792 continue;
793 }
794 if (done >= prev_done) {
795 // Newly completed — print immediately
796 try stdout.print(
797 "[{d}] completed task={s} exit_code={d}\n",
798 .{ session.task_ended_at.?, session.name, session.task_exit_code.? },
799 );
800 try stdout.flush();
801 }
802 if (session.task_exit_code != 0) {
803 agg_exit_code = session.task_exit_code orelse 0;
804 }
805 done += 1;
806 }
807
808 for (sessions.items) |session| {
809 session.deinit(alloc);
810 }
811 sessions.deinit(alloc);
812
813 // Check disappearance BEFORE completion: if one of N sessions
814 // crashed and the remaining N-1 happen to be done, total==done
815 // would be a false success.
816 if (total < max_seen) {
817 try stderr.print(
818 "error: {d} session(s) disappeared before completing\n",
819 .{max_seen - total},
820 );
821 try stderr.flush();
822 std.process.exit(1);
823 return;
824 }
825 max_seen = total;
826
827 if (total > 0 and total == done) {
828 break;
829 }
830
831 if (max_seen == 0) {
832 // `zmx run foo && zmx wait foo` is essentially sequential, so
833 // matching sessions should be visible from the first poll. If
834 // nothing appears after a few iterations it's almost certainly a
835 // typo, not a slow start.
836 zero_match_iters += 1;
837 if (zero_match_iters >= 3) {
838 try stderr.print("error: no matching sessions found\n", .{});
839 try stderr.flush();
840 std.process.exit(2);
841 return;
842 }
843 }
844
845 prev_done = done;
846 std.Io.sleep(io, std.Io.Duration.fromMilliseconds(1000), .real) catch unreachable;
847 }
848
849 if (agg_exit_code == 0) {
850 try stdout.print("task(s) completed!\n", .{});
851 } else {
852 try stdout.print("task(s) failed!\n", .{});
853 }
854 try stdout.flush();
855
856 const sessions = try util.get_session_entries(alloc, io, cfg.socket_dir);
857 for (sessions.items) |session| {
858 var found = false;
859 for (matchers.items) |m| {
860 if (m.matches(session.name)) {
861 found = true;
862 break;
863 }
864 }
865 if (!found) {
866 continue;
867 }
868 if (session.task_exit_code.? > 0) {
869 try stdout.print("---\n", .{});
870 try stdout.print("[{d}] failed task={s} exit_status={d}\n", .{
871 session.task_ended_at.?,
872 session.name,
873 session.task_exit_code.?,
874 });
875
876 // Fetch and print the last 20 lines of history for debugging
877 const history_lines: usize = 20;
878 const history_text = fetchHistory(alloc, io, cfg, session.name) catch null;
879 if (history_text) |text| {
880 defer alloc.free(text);
881 try stdout.print("\nLast {d} lines of {s} history:\n", .{ history_lines, session.name });
882
883 // Count lines and find the start of the last N lines
884 var total_lines: usize = 0;
885 var it = std.mem.splitScalar(u8, text, '\n');
886 while (it.next()) |_| {
887 total_lines += 1;
888 }
889
890 const skip = if (total_lines > history_lines) total_lines - history_lines else 0;
891 var current: usize = 0;
892 it = std.mem.splitScalar(u8, text, '\n');
893 while (it.next()) |line| {
894 if (current >= skip) {
895 try stdout.print("{s}\n", .{line});
896 }
897 current += 1;
898 }
899 }
900
901 try stdout.print("\nSee the logs:\nzmx history {s}\nzmx attach {s}\n", .{ session.name, session.name });
902 try stdout.flush();
903 }
904 }
905
906 std.process.exit(agg_exit_code);
907}
908
909fn list(alloc: std.mem.Allocator, io: std.Io, cfg: *Cfg, short: bool) !void {
910 const current_session = socket.getSeshNameFromEnv();
911 var buf: [4096]u8 = undefined;
912 var stdout = std.Io.File.stdout().writer(io, &buf);
913 var sessions = try util.get_session_entries(alloc, io, cfg.socket_dir);
914 defer {
915 for (sessions.items) |session| {
916 session.deinit(alloc);
917 }
918 sessions.deinit(alloc);
919 }
920
921 if (sessions.items.len == 0) {
922 if (short) return;
923 var errbuf: [4096]u8 = undefined;
924 var stderr = std.Io.File.stderr().writer(io, &errbuf);
925 try stderr.interface.print("no sessions found in {s}\n", .{cfg.socket_dir});
926 try stderr.interface.flush();
927 return;
928 }
929
930 std.mem.sort(util.SessionEntry, sessions.items, {}, util.SessionEntry.lessThan);
931
932 for (sessions.items) |session| {
933 if (session.is_error) {
934 try util.writeSessionLine(&stdout.interface, session, short, current_session);
935 try stdout.interface.flush();
936 continue;
937 }
938
939 try util.writeSessionLine(&stdout.interface, session, short, current_session);
940 try stdout.interface.flush();
941 }
942}
943
944fn detachAll(alloc: std.mem.Allocator, io: std.Io, cfg: *Cfg) !void {
945 const session_name = socket.getSeshNameFromEnv();
946 if (session_name.len == 0) {
947 std.log.err("ZMX_SESSION env var not found: are you inside a zmx session?", .{});
948 return;
949 }
950 std.log.info("detach all session={s}", .{session_name});
951
952 var dir = try std.Io.Dir.openDirAbsolute(io, cfg.socket_dir, .{});
953 defer dir.close(io);
954
955 const socket_path = socket.getSocketPath(alloc, cfg.socket_dir, session_name) catch |err| switch (err) {
956 error.NameTooLong => return socket.printSessionNameTooLong(io, session_name, cfg.socket_dir),
957 error.OutOfMemory => return err,
958 };
959 defer alloc.free(socket_path);
960 const fd = ipc.connectSession(socket_path) catch |err| {
961 std.log.err("session unresponsive: {s}", .{@errorName(err)});
962 if (err == error.ConnectionRefused) socket.cleanupStaleSocket(io, dir, session_name);
963 return;
964 };
965 defer lib_posix.close(fd);
966 ipc.send(fd, .DetachAll, "") catch |err| switch (err) {
967 error.BrokenPipe, error.ConnectionResetByPeer => return,
968 else => return err,
969 };
970}
971
972fn kill(alloc: std.mem.Allocator, io: std.Io, cfg: *Cfg, session_name: []const u8, force: bool) !void {
973 std.log.info("kill session={s}", .{session_name});
974 const socket_path = socket.getSocketPath(alloc, cfg.socket_dir, session_name) catch |err| switch (err) {
975 error.NameTooLong => return socket.printSessionNameTooLong(io, session_name, cfg.socket_dir),
976 error.OutOfMemory => return err,
977 };
978 defer alloc.free(socket_path);
979
980 var dir = try std.Io.Dir.openDirAbsolute(io, cfg.socket_dir, .{});
981 defer dir.close(io);
982
983 const exists = try socket.sessionExists(io, dir, session_name);
984 if (!exists) {
985 var buf: [4096]u8 = undefined;
986 var w = std.Io.File.stderr().writer(io, &buf);
987 w.interface.print("error: session \"{s}\" does not exist\n", .{session_name}) catch {};
988 w.interface.flush() catch {};
989 return error.SessionNotFound;
990 }
991 const fd = ipc.connectSession(socket_path) catch |err| {
992 std.log.err("session unresponsive: {s}", .{@errorName(err)});
993 var buf: [4096]u8 = undefined;
994 var w = std.Io.File.stdout().writer(io, &buf);
995 if (force or err == error.ConnectionRefused) {
996 socket.cleanupStaleSocket(io, dir, session_name);
997 w.interface.print("cleaned up stale session {s}\n", .{session_name}) catch {};
998 } else {
999 w.interface.print(
1000 "session {s} is unresponsive ({s})\ndaemon may be busy: try again, add `--force` flag, or kill the process directly\n",
1001 .{ session_name, @errorName(err) },
1002 ) catch {};
1003 }
1004 w.interface.flush() catch {};
1005 return;
1006 };
1007
1008 defer lib_posix.close(fd);
1009 ipc.send(fd, .Kill, "") catch |err| switch (err) {
1010 error.BrokenPipe, error.ConnectionResetByPeer => return,
1011 else => return err,
1012 };
1013
1014 // Block until the daemon hangs up. The daemon's shutdown defer closes
1015 // and unlinks the listen socket before it closes client connections,
1016 // so by the time we read EOF here the session name is free for reuse
1017 // and a subsequent `zmx run <name>` can't land in the dying daemon's
1018 // accept backlog.
1019 var drain: [256]u8 = undefined;
1020 while (true) {
1021 const n = lib_posix.read(fd, &drain) catch break;
1022 if (n == 0) break;
1023 }
1024
1025 var buf: [100]u8 = undefined;
1026 var w = std.Io.File.stdout().writer(io, &buf);
1027 try w.interface.print("killed session {s}\n", .{session_name});
1028 try w.interface.flush();
1029}
1030
1031fn printLabelError(io: std.Io, session_name: []const u8, err: anyerror) noreturn {
1032 var buf: [4096]u8 = undefined;
1033 var w = std.Io.File.stderr().writer(io, &buf);
1034 switch (err) {
1035 error.Timeout => w.interface.print(
1036 "error: session \"{s}\" does not support labels (daemon too old?)\n",
1037 .{session_name},
1038 ) catch {},
1039 error.ConnectionRefused, error.Unexpected => w.interface.print(
1040 "error: session \"{s}\" not found or unresponsive\n",
1041 .{session_name},
1042 ) catch {},
1043 else => w.interface.print(
1044 "error: {s}\n",
1045 .{@errorName(err)},
1046 ) catch {},
1047 }
1048 w.interface.flush() catch {};
1049 std.process.exit(1);
1050}
1051
1052fn labelGet(alloc: std.mem.Allocator, io: std.Io, cfg: *Cfg, session_name: []const u8, single_kv: []const u8) !void {
1053 std.log.info("label get session={s}", .{session_name});
1054
1055 const socket_path = socket.getSocketPath(alloc, cfg.socket_dir, session_name) catch |err| switch (err) {
1056 error.NameTooLong => return socket.printSessionNameTooLong(io, session_name, cfg.socket_dir),
1057 error.OutOfMemory => return err,
1058 };
1059 defer alloc.free(socket_path);
1060
1061 const payload = ipc.roundTripForTag(alloc, socket_path, .LabelGet, "", .LabelData) catch |err| {
1062 printLabelError(io, session_name, err);
1063 };
1064 defer alloc.free(payload);
1065
1066 var buf: [4096]u8 = undefined;
1067 var stdout = std.Io.File.stdout().writer(io, &buf);
1068 if (single_kv.len == 0) {
1069 try stdout.interface.print("{s}", .{payload});
1070 try stdout.interface.flush();
1071 return;
1072 }
1073
1074 const val = try label.getLabelValueFromPairs(single_kv, payload);
1075 try stdout.interface.print("{s}", .{val});
1076 try stdout.interface.flush();
1077}
1078
1079fn labelSet(alloc: std.mem.Allocator, io: std.Io, cfg: *Cfg, session_name: []const u8, labels: []const u8) !void {
1080 std.log.info("label set session={s}", .{session_name});
1081
1082 var kvs = label.LabelIterator.init(labels);
1083 while (kvs.next()) |kv| {
1084 label.assertLabel(kv.key, kv.value) catch |err| {
1085 var buf: [4096]u8 = undefined;
1086 var w = std.Io.File.stderr().writer(io, &buf);
1087 const msg = "error: key-value kvs can only contain [a-z, A-Z, 0-9, -_.] characters";
1088 switch (err) {
1089 error.LabelKeyEmpty => {
1090 w.interface.print("error: label key cannot be empty\n", .{}) catch {};
1091 },
1092 error.LabelKeyReservedName => {
1093 w.interface.print("error: \"{s}\" is a read-only built-in field\n", .{kv.key}) catch {};
1094 },
1095 error.LabelKeyInvalidChar => {
1096 w.interface.print("{s}: key=[{s}]\n", .{ msg, kv.key }) catch {};
1097 },
1098 error.LabelValueInvalidChar => {
1099 w.interface.print("{s}: value=[{s}]\n", .{ msg, kv.value }) catch {};
1100 },
1101 }
1102 w.interface.flush() catch {};
1103 std.process.exit(1);
1104 };
1105 }
1106
1107 const socket_path = socket.getSocketPath(alloc, cfg.socket_dir, session_name) catch |err| switch (err) {
1108 error.NameTooLong => return socket.printSessionNameTooLong(io, session_name, cfg.socket_dir),
1109 error.OutOfMemory => return err,
1110 };
1111 defer alloc.free(socket_path);
1112
1113 _ = ipc.roundTripForTag(alloc, socket_path, .LabelSet, labels, .Ack) catch |err| {
1114 printLabelError(io, session_name, err);
1115 };
1116}
1117
1118fn labelClear(alloc: std.mem.Allocator, io: std.Io, cfg: *Cfg, session_name: []const u8) !void {
1119 std.log.info("label clear session={s}", .{session_name});
1120
1121 const socket_path = socket.getSocketPath(alloc, cfg.socket_dir, session_name) catch |err| switch (err) {
1122 error.NameTooLong => return socket.printSessionNameTooLong(io, session_name, cfg.socket_dir),
1123 error.OutOfMemory => return err,
1124 };
1125 defer alloc.free(socket_path);
1126
1127 _ = ipc.roundTripForTag(alloc, socket_path, .LabelClear, "", .Ack) catch |err| {
1128 printLabelError(io, session_name, err);
1129 };
1130}
1131
1132/// Fetch terminal history from a session socket, returning it as an allocated
1133/// string. Caller owns the returned memory and must free it.
1134fn fetchHistory(
1135 alloc: std.mem.Allocator,
1136 io: std.Io,
1137 cfg: *Cfg,
1138 session_name: []const u8,
1139) ![]const u8 {
1140 std.log.info("fetch history session={s}", .{session_name});
1141 const socket_path = socket.getSocketPath(alloc, cfg.socket_dir, session_name) catch |err| switch (err) {
1142 error.NameTooLong => {
1143 socket.printSessionNameTooLong(io, session_name, cfg.socket_dir);
1144 return error.NameTooLong;
1145 },
1146 error.OutOfMemory => return err,
1147 };
1148 defer alloc.free(socket_path);
1149
1150 var dir = try std.Io.Dir.openDirAbsolute(io, cfg.socket_dir, .{});
1151 defer dir.close(io);
1152
1153 const exists = try socket.sessionExists(io, dir, session_name);
1154 if (!exists) {
1155 return error.SessionNotFound;
1156 }
1157
1158 const fd = ipc.connectSession(socket_path) catch |err| {
1159 if (err == error.ConnectionRefused) socket.cleanupStaleSocket(io, dir, session_name);
1160 return err;
1161 };
1162 defer lib_posix.close(fd);
1163
1164 const format_byte: u8 = @intFromEnum(util.HistoryFormat.plain);
1165 const payload = [_]u8{format_byte};
1166 ipc.send(fd, .History, &payload) catch |err| switch (err) {
1167 error.BrokenPipe, error.ConnectionResetByPeer => return error.SessionUnresponsive,
1168 else => return err,
1169 };
1170
1171 var sb = try ipc.SocketBuffer.init(alloc);
1172 defer sb.deinit();
1173
1174 var result = std.ArrayList(u8).initCapacity(alloc, 4096) catch return error.OutOfMemory;
1175 errdefer result.deinit(alloc);
1176
1177 while (true) {
1178 var poll_fds = [_]lib_posix.pollfd{.{ .fd = fd, .events = lib_posix.POLL.IN, .revents = 0 }};
1179 const poll_result = lib_posix.poll(&poll_fds, 5000) catch return error.Timeout;
1180 if (poll_result == 0) {
1181 return error.Timeout;
1182 }
1183
1184 const n = sb.read(fd) catch return error.ReadFailed;
1185 if (n == 0) break;
1186
1187 while (sb.next()) |msg| {
1188 if (msg.header.tag == .History) {
1189 try result.appendSlice(alloc, msg.payload);
1190 return result.toOwnedSlice(alloc);
1191 }
1192 }
1193 }
1194
1195 return error.NoHistoryResponse;
1196}
1197
1198fn history(alloc: std.mem.Allocator, io: std.Io, cfg: *Cfg, session_name: []const u8, format: util.HistoryFormat) !void {
1199 std.log.info("history session={s}", .{session_name});
1200
1201 const socket_path = socket.getSocketPath(alloc, cfg.socket_dir, session_name) catch |err| switch (err) {
1202 error.NameTooLong => return socket.printSessionNameTooLong(io, session_name, cfg.socket_dir),
1203 error.OutOfMemory => return err,
1204 };
1205 defer alloc.free(socket_path);
1206
1207 var dir = try std.Io.Dir.openDirAbsolute(io, cfg.socket_dir, .{});
1208 defer dir.close(io);
1209
1210 const exists = try socket.sessionExists(io, dir, session_name);
1211 if (!exists) {
1212 var buf: [4096]u8 = undefined;
1213 var w = std.Io.File.stderr().writer(io, &buf);
1214 w.interface.print("error: session \"{s}\" does not exist\n", .{session_name}) catch {};
1215 w.interface.flush() catch {};
1216 return error.SessionNotFound;
1217 }
1218 const fd = ipc.connectSession(socket_path) catch |err| {
1219 std.log.err("session unresponsive: {s}", .{@errorName(err)});
1220 if (err == error.ConnectionRefused) socket.cleanupStaleSocket(io, dir, session_name);
1221 return;
1222 };
1223 defer lib_posix.close(fd);
1224
1225 const format_byte = [_]u8{@intFromEnum(format)};
1226 ipc.send(fd, .History, &format_byte) catch |err| switch (err) {
1227 error.BrokenPipe, error.ConnectionResetByPeer => return,
1228 else => return err,
1229 };
1230
1231 var sb = try ipc.SocketBuffer.init(alloc);
1232 defer sb.deinit();
1233
1234 while (true) {
1235 var poll_fds = [_]lib_posix.pollfd{.{ .fd = fd, .events = lib_posix.POLL.IN, .revents = 0 }};
1236 const poll_result = lib_posix.poll(&poll_fds, 5000) catch return;
1237 if (poll_result == 0) {
1238 std.log.err("timeout waiting for history response", .{});
1239 return;
1240 }
1241
1242 const n = sb.read(fd) catch return;
1243 if (n == 0) return;
1244
1245 while (sb.next()) |msg| {
1246 if (msg.header.tag == .History) {
1247 _ = lib_posix.write(lib_posix.STDOUT_FILENO, msg.payload) catch return;
1248 return;
1249 }
1250 }
1251 }
1252}
1253
1254fn switchSesh(gpa: std.mem.Allocator, io: std.Io, daemon: *Daemon, current_sesh: []const u8) !void {
1255 // we want daemon.session_name because that's the session name the user provided during zmx attach
1256 // instead of the name of the session they are currently inside of.
1257 const next_session = daemon.session_name;
1258 std.log.info("switch session cur={s} next={s}", .{ current_sesh, next_session });
1259
1260 const socket_path = socket.getSocketPath(gpa, daemon.cfg.socket_dir, current_sesh) catch |err| switch (err) {
1261 error.NameTooLong => return socket.printSessionNameTooLong(io, current_sesh, daemon.cfg.socket_dir),
1262 error.OutOfMemory => return err,
1263 };
1264 defer gpa.free(socket_path);
1265
1266 var dir = try std.Io.Dir.openDirAbsolute(io, daemon.cfg.socket_dir, .{});
1267 defer dir.close(io);
1268
1269 const exists = try socket.sessionExists(io, dir, current_sesh);
1270 if (!exists) {
1271 var buf: [4096]u8 = undefined;
1272 var w = std.Io.File.stderr().writer(io, &buf);
1273 w.interface.print("error: session \"{s}\" does not exist\n", .{current_sesh}) catch {};
1274 w.interface.flush() catch {};
1275 return error.SessionNotFound;
1276 }
1277 const fd = ipc.connectSession(socket_path) catch |err| {
1278 std.log.err("session unresponsive: {s}", .{@errorName(err)});
1279 if (err == error.ConnectionRefused) socket.cleanupStaleSocket(io, dir, current_sesh);
1280 return;
1281 };
1282 defer lib_posix.close(fd);
1283
1284 ipc.send(fd, .Switch, next_session) catch |err| switch (err) {
1285 error.BrokenPipe, error.ConnectionResetByPeer => return,
1286 else => return err,
1287 };
1288}
1289
1290fn attach(gpa: std.mem.Allocator, io: std.Io, daemon: *Daemon) !void {
1291 const sesh = socket.getSeshNameFromEnv();
1292 if (sesh.len > 0) {
1293 return switchSesh(gpa, io, daemon, sesh);
1294 }
1295
1296 const is_daemon_proc = try daemon.ensureSession(io);
1297 if (is_daemon_proc) return;
1298
1299 const client_sock = try socket.sessionConnect(daemon.socket_path);
1300 std.log.info("attached session={s}", .{daemon.session_name});
1301 // This is typically used with tcsetattr() to modify terminal settings.
1302 // - you first get the current settings with tcgetattr()
1303 // - modify the desired attributes in the termios structure
1304 // - then apply the changes with tcsetattr().
1305 // This prevents unintended side effects by preserving other settings.
1306 // restore stdin fd to its original state after exiting.
1307 // Use TCSAFLUSH to discard any unread input, preventing stale input after detach.
1308 //
1309 // tcgetattr fails when stdin is not a TTY (e.g. piped). In that case,
1310 // skip terminal setup entirely rather than applying undefined stack bytes
1311 // via tcsetattr.
1312 var orig_termios: cross.c.termios = undefined;
1313 const stdin_is_tty = cross.c.tcgetattr(lib_posix.STDIN_FILENO, &orig_termios) == 0;
1314
1315 defer {
1316 if (stdin_is_tty) {
1317 _ = cross.c.tcsetattr(lib_posix.STDIN_FILENO, cross.c.TCSAFLUSH, &orig_termios);
1318 }
1319 // Reset terminal modes on detach
1320 const restore_seq = "\x1bc";
1321 _ = lib_posix.write(lib_posix.STDOUT_FILENO, restore_seq) catch {};
1322 }
1323
1324 if (stdin_is_tty) {
1325 var raw_termios = orig_termios;
1326 // set raw mode after successful connection.
1327 // disables canonical mode (line buffering), input echoing, signal generation from
1328 // control characters (like Ctrl+C), and flow control.
1329 cross.c.cfmakeraw(&raw_termios);
1330
1331 // Additional granular raw mode settings for precise control
1332 // (matches what abduco and shpool do)
1333 raw_termios.c_cc[cross.c.VLNEXT] = cross.c._POSIX_VDISABLE; // Disable literal-next (Ctrl-V)
1334 // We want to intercept Ctrl+\ (SIGQUIT) so we can use it as a detach key
1335 raw_termios.c_cc[cross.c.VQUIT] = cross.c._POSIX_VDISABLE; // Disable SIGQUIT (Ctrl+\)
1336 raw_termios.c_cc[cross.c.VMIN] = 1; // Minimum chars to read: return after 1 byte
1337 raw_termios.c_cc[cross.c.VTIME] = 0; // Read timeout: no timeout, return immediately
1338
1339 _ = cross.c.tcsetattr(lib_posix.STDIN_FILENO, cross.c.TCSANOW, &raw_termios);
1340 }
1341
1342 // Clear screen before attaching. This provides a clean slate before
1343 // the session restore.
1344 const clear_seq = "\x1b[2J\x1b[H";
1345 _ = try lib_posix.write(lib_posix.STDOUT_FILENO, clear_seq);
1346
1347 const looper = try loop.clientLoop(client_sock);
1348 switch (looper.kind) {
1349 .detach => return,
1350 .switch_session => {
1351 if (looper.session_name) |session_name| {
1352 // Reset terminal modes when switching sessions
1353 const restore_seq = "\x1bc";
1354 _ = lib_posix.write(lib_posix.STDOUT_FILENO, restore_seq) catch {};
1355
1356 const target_path = socket.getSocketPath(
1357 gpa,
1358 daemon.cfg.socket_dir,
1359 session_name,
1360 ) catch |err| switch (err) {
1361 error.NameTooLong => return socket.printSessionNameTooLong(
1362 io,
1363 session_name,
1364 daemon.cfg.socket_dir,
1365 ),
1366 error.OutOfMemory => return err,
1367 };
1368
1369 var target_daemon = Daemon.init(io, daemon.cfg, session_name, target_path);
1370 // Use the cwd from the previous daemon if available (sent by the daemon),
1371 // otherwise fall back to the client's original cwd
1372 const switch_cwd = looper.cwd orelse daemon.cwd;
1373 std.log.info("switching to new session cwd={s}", .{switch_cwd});
1374 target_daemon.setCwd(switch_cwd);
1375 target_daemon.shell = daemon.shell;
1376 return attach(gpa, io, &target_daemon);
1377 }
1378 },
1379 }
1380}
1381
1382fn writeFile(gpa: std.mem.Allocator, io: std.Io, daemon: *Daemon, file_path: []const u8) !void {
1383 const is_daemon_proc = try daemon.ensureSession(io);
1384 if (is_daemon_proc) return;
1385
1386 var buf: [4096]u8 = undefined;
1387 var w = std.Io.File.stdout().writer(io, &buf);
1388
1389 const stdin_fd = lib_posix.STDIN_FILENO;
1390 var stdin_buf = try std.ArrayList(u8).initCapacity(gpa, 4096);
1391 defer stdin_buf.deinit(gpa);
1392
1393 while (true) {
1394 var tmp: [4096]u8 = undefined;
1395 const n = lib_posix.read(stdin_fd, &tmp) catch |err| {
1396 if (err == error.WouldBlock) break;
1397 return err;
1398 };
1399 if (n == 0) break;
1400 try stdin_buf.appendSlice(gpa, tmp[0..n]);
1401 }
1402
1403 const socket_path = socket.getSocketPath(
1404 gpa,
1405 daemon.cfg.socket_dir,
1406 daemon.session_name,
1407 ) catch |err| switch (err) {
1408 error.NameTooLong => return socket.printSessionNameTooLong(
1409 io,
1410 daemon.session_name,
1411 daemon.cfg.socket_dir,
1412 ),
1413 error.OutOfMemory => return err,
1414 };
1415 var dir = try std.Io.Dir.openDirAbsolute(io, daemon.cfg.socket_dir, .{});
1416 defer dir.close(io);
1417
1418 const result = ipc.probeSession(gpa, socket_path) catch |err| {
1419 std.log.err("session unresponsive: {s}", .{@errorName(err)});
1420 if (err == error.ConnectionRefused) {
1421 socket.cleanupStaleSocket(io, dir, daemon.session_name);
1422 w.interface.print("cleaned up stale session {s}\n", .{daemon.session_name}) catch {};
1423 } else {
1424 w.interface.print(
1425 "session {s} is unresponsive ({s})\ndaemon may be busy: try again\n",
1426 .{ daemon.session_name, @errorName(err) },
1427 ) catch {};
1428 }
1429 w.interface.flush() catch {};
1430 return;
1431 };
1432
1433 defer result.deinit();
1434
1435 // Build wire payload: [u32 path len][path bytes][file content]
1436 var wire_buf = try std.ArrayList(u8).initCapacity(
1437 gpa,
1438 @sizeOf(u32) + file_path.len + stdin_buf.items.len,
1439 );
1440 defer wire_buf.deinit(gpa);
1441 const path_len: u32 = @intCast(file_path.len);
1442 try wire_buf.appendSlice(gpa, std.mem.asBytes(&path_len));
1443 try wire_buf.appendSlice(gpa, file_path);
1444 try wire_buf.appendSlice(gpa, stdin_buf.items);
1445
1446 ipc.send(result.fd, .Write, wire_buf.items) catch |err| switch (err) {
1447 error.BrokenPipe, error.ConnectionResetByPeer => return,
1448 else => return err,
1449 };
1450
1451 var sb = try ipc.SocketBuffer.init(gpa);
1452 defer sb.deinit();
1453
1454 const n = sb.read(result.fd) catch return error.ReadFailed;
1455 if (n == 0) return error.ConnectionClosed;
1456
1457 while (sb.next()) |msg| {
1458 if (msg.header.tag == .Ack) {
1459 try w.interface.print("file created {s}\n", .{file_path});
1460 try w.interface.flush();
1461 return;
1462 }
1463 }
1464
1465 return error.NoAckReceived;
1466}
1467
1468fn send(alloc: std.mem.Allocator, io: std.Io, cfg: *Cfg, session_name: []const u8, socket_path: []const u8, text_parts: [][]const u8, tag: ipc.Tag) !void {
1469 std.log.info("send session={s}", .{session_name});
1470 var buf: [4096]u8 = undefined;
1471 var w = std.Io.File.stdout().writer(io, &buf);
1472
1473 var payload = std.ArrayList(u8).empty;
1474 defer payload.deinit(alloc);
1475
1476 if (text_parts.len > 0) {
1477 for (text_parts, 0..) |part, i| {
1478 if (i > 0) try payload.append(alloc, ' ');
1479 try payload.appendSlice(alloc, part);
1480 }
1481 } else {
1482 // Read from stdin when no text arguments provided.
1483 const stdin_file = std.Io.File.stdin();
1484 defer stdin_file.close(io);
1485 var stdin_buf: [4096]u8 = undefined;
1486 var reader = stdin_file.reader(io, &stdin_buf);
1487 if (!try stdin_file.isTty(io)) {
1488 while (true) {
1489 var dest: [1024]u8 = undefined;
1490 const n = try reader.interface.readSliceShort(&dest);
1491 if (n == 0) break; // EOF
1492 try payload.appendSlice(alloc, dest[0..n]);
1493 }
1494 // Strip trailing newline from piped input; the caller is
1495 // responsible for including \r when submission is desired.
1496 // For .Output the caller controls exact bytes, so don't strip.
1497 if (tag != .Output and payload.items.len > 0 and payload.items[payload.items.len - 1] == '\n') {
1498 _ = payload.pop();
1499 }
1500 }
1501 }
1502
1503 if (payload.items.len == 0) return error.TextRequired;
1504
1505 var dir = try std.Io.Dir.openDirAbsolute(io, cfg.socket_dir, .{});
1506 defer dir.close(io);
1507
1508 const probe_result = ipc.probeSession(alloc, socket_path) catch |err| {
1509 std.log.err("session unresponsive: {s}", .{@errorName(err)});
1510 if (err == error.ConnectionRefused) {
1511 socket.cleanupStaleSocket(io, dir, session_name);
1512 try w.interface.print("cleaned up stale session {s}\n", .{session_name});
1513 } else {
1514 try w.interface.print(
1515 "session {s} is unresponsive ({s})\ndaemon may be busy: try again\n",
1516 .{ session_name, @errorName(err) },
1517 );
1518 }
1519 try w.interface.flush();
1520 return;
1521 };
1522 defer probe_result.deinit();
1523
1524 ipc.send(probe_result.fd, tag, payload.items) catch |err| switch (err) {
1525 error.ConnectionResetByPeer, error.BrokenPipe => return,
1526 else => return err,
1527 };
1528}
1529
1530fn run(gpa: std.mem.Allocator, io: std.Io, daemon: *Daemon, detached: bool, command_args: [][]const u8) !void {
1531 var cmd_to_send: ?[]const u8 = null;
1532 var allocated_cmd: ?[]u8 = null;
1533 defer if (allocated_cmd) |cmd| gpa.free(cmd);
1534
1535 const is_daemon_proc = try daemon.ensureSession(io);
1536 if (is_daemon_proc) return;
1537
1538 if (command_args.len > 0) {
1539 var cmd_list = std.ArrayList(u8).empty;
1540 defer cmd_list.deinit(gpa);
1541
1542 for (command_args, 0..) |arg, i| {
1543 if (i > 0) try cmd_list.append(gpa, ' ');
1544 if (util.shellNeedsQuoting(arg)) {
1545 const quoted = try util.shellQuote(gpa, arg);
1546 defer gpa.free(quoted);
1547 try cmd_list.appendSlice(gpa, quoted);
1548 } else {
1549 try cmd_list.appendSlice(gpa, arg);
1550 }
1551 }
1552
1553 // \r, not \n: once the shell is at the readline prompt the PTY is in
1554 // raw mode; readline's accept-line binds to CR. The first-ever run
1555 // works with \n only because it arrives during shell startup while
1556 // the line discipline is still canonical.
1557 try cmd_list.append(gpa, '\r');
1558
1559 cmd_to_send = try cmd_list.toOwnedSlice(gpa);
1560 allocated_cmd = @constCast(cmd_to_send.?);
1561 } else {
1562 // Read from stdin when no text arguments provided.
1563 const stdin_file = std.Io.File.stdin();
1564 defer stdin_file.close(io);
1565 var stdin_buf = try std.ArrayList(u8).initCapacity(gpa, 4096);
1566 defer stdin_buf.deinit(gpa);
1567 var stdbuf: [4096]u8 = undefined;
1568 var reader = stdin_file.reader(io, &stdbuf);
1569 if (!try stdin_file.isTty(io)) {
1570 while (true) {
1571 var dest: [1024]u8 = undefined;
1572 const n = try reader.interface.readSliceShort(&dest);
1573 if (n == 0) break; // EOF
1574 try stdin_buf.appendSlice(gpa, dest[0..n]);
1575 }
1576
1577 if (stdin_buf.items.len > 0) {
1578 // Normalize any trailing newline to CR so readline (raw mode)
1579 // accepts each line.
1580 if (stdin_buf.items[stdin_buf.items.len - 1] == '\n') {
1581 stdin_buf.items[stdin_buf.items.len - 1] = '\r';
1582 } else {
1583 try stdin_buf.append(gpa, '\r');
1584 }
1585
1586 cmd_to_send = try gpa.dupe(u8, stdin_buf.items);
1587 allocated_cmd = @constCast(cmd_to_send.?);
1588 }
1589 }
1590 }
1591
1592 if (cmd_to_send == null) {
1593 return error.CommandRequired;
1594 }
1595
1596 const client_sock = ipc.connectSession(daemon.socket_path) catch |err| {
1597 std.log.err("session not ready: {s}", .{@errorName(err)});
1598 return error.SessionNotReady;
1599 };
1600 defer lib_posix.close(client_sock);
1601
1602 const term_size = ipc.getTerminalSize(lib_posix.STDOUT_FILENO);
1603 ipc.send(client_sock, .Resize, std.mem.asBytes(&term_size)) catch {};
1604
1605 var fds = try std.ArrayList(i32).initCapacity(gpa, 1);
1606 defer fds.deinit(gpa);
1607 try fds.append(gpa, client_sock);
1608
1609 ipc.send(client_sock, .Run, cmd_to_send.?) catch |err| switch (err) {
1610 error.ConnectionResetByPeer, error.BrokenPipe => return,
1611 else => return err,
1612 };
1613
1614 const exit_code = try tail(gpa, fds, detached, true);
1615 lib_posix.exit(exit_code);
1616}