Barrett Ruth
·
2026-10-26
1const std = @import("std");
2const ipc = @import("ipc.zig");
3const log = @import("log.zig");
4const util = @import("util.zig");
5const cross = @import("cross");
6const socket = @import("socket.zig");
7const label = @import("label.zig");
8const lib_posix = @import("posix.zig");
9const Cfg = @import("cfg.zig");
10const signal = @import("signal.zig");
11const assert = std.debug.assert;
12const daemonize = @import("daemonize.zig");
13const builtin = @import("builtin");
14const terminal = @import("terminal.zig");
15const ghostty_vt = terminal.ghostty_vt;
16
17/// clientLoop sends ipc commands to its corresponding daemon. It uses poll() as its non-blocking
18/// mechanism. It will send stdin to the daemon and receive stdout from the daemon.
19pub fn clientLoop(client_sock_fd: i32, env_str: []const u8) !ClientResult {
20 std.log.info("client loop fd={d}", .{client_sock_fd});
21 const gpa: std.mem.Allocator = blk: {
22 if (builtin.mode == .debug) {
23 const GPA = std.heap.DebugAllocator(.{});
24 const Static = struct {
25 var gpa: GPA = .{};
26 };
27 break :blk Static.gpa.allocator();
28 }
29 break :blk std.heap.c_allocator;
30 };
31 defer lib_posix.close(client_sock_fd);
32
33 try signal.openSignalPipe();
34 defer signal.closeSignalPipe();
35 signal.installWakeHandler(@backingInt(lib_posix.SIG.WINCH));
36
37 // Make socket non-blocking to avoid blocking on writes
38 var sock_flags = try lib_posix.fcntl(client_sock_fd, lib_posix.F.GETFL, 0);
39 sock_flags |= lib_posix.O_NONBLOCK;
40 _ = try lib_posix.fcntl(client_sock_fd, lib_posix.F.SETFL, sock_flags);
41
42 // Buffer for outgoing socket writes
43 var sock_write_buf = try std.ArrayList(u8).initCapacity(gpa, 4096);
44 defer sock_write_buf.deinit(gpa);
45
46 if (env_str.len > 0) {
47 try ipc.appendMessage(gpa, &sock_write_buf, .EnvSet, env_str);
48 }
49
50 // Send init message with terminal size (buffered)
51 const size = ipc.getTerminalSize(lib_posix.STDOUT_FILENO);
52 try ipc.appendMessage(gpa, &sock_write_buf, .Init, std.mem.asBytes(&size));
53
54 var poll_fds = try std.ArrayList(lib_posix.pollfd).initCapacity(gpa, 4);
55 defer poll_fds.deinit(gpa);
56
57 var read_buf = try ipc.SocketBuffer.init(gpa);
58 defer read_buf.deinit();
59
60 var stdout_buf = try std.ArrayList(u8).initCapacity(gpa, 4096);
61 defer stdout_buf.deinit(gpa);
62
63 const stdin_fd = lib_posix.STDIN_FILENO;
64
65 // Make stdin non-blocking. O_NONBLOCK is set on the open file description,
66 // which is shared with the parent shell; restore on exit to avoid
67 // corrupting the parent's stdin.
68 const stdin_orig_flags = try lib_posix.fcntl(stdin_fd, lib_posix.F.GETFL, 0);
69 _ = try lib_posix.fcntl(stdin_fd, lib_posix.F.SETFL, stdin_orig_flags | lib_posix.O_NONBLOCK);
70 defer _ = lib_posix.fcntl(stdin_fd, lib_posix.F.SETFL, stdin_orig_flags) catch {};
71
72 const detach_key_disabled = util.isDetachKeyDisabled();
73
74 while (true) {
75 poll_fds.clearRetainingCapacity();
76
77 try poll_fds.append(gpa, .{
78 .fd = stdin_fd,
79 .events = lib_posix.POLL.IN,
80 .revents = 0,
81 });
82
83 // Poll socket for read, and also for write if we have pending data
84 var sock_events: i16 = lib_posix.POLL.IN;
85 if (sock_write_buf.items.len > 0) {
86 sock_events |= lib_posix.POLL.OUT;
87 }
88 try poll_fds.append(gpa, .{
89 .fd = client_sock_fd,
90 .events = sock_events,
91 .revents = 0,
92 });
93
94 try poll_fds.append(gpa, .{ .fd = signal.sig_pipe[0], .events = lib_posix.POLL.IN, .revents = 0 });
95
96 if (stdout_buf.items.len > 0) {
97 try poll_fds.append(gpa, .{
98 .fd = lib_posix.STDOUT_FILENO,
99 .events = lib_posix.POLL.OUT,
100 .revents = 0,
101 });
102 }
103
104 _ = try lib_posix.poll(poll_fds.items, -1);
105
106 if (poll_fds.items[2].revents & lib_posix.POLL.IN != 0) {
107 signal.drainSignalPipe();
108 const next_size = ipc.getTerminalSize(lib_posix.STDOUT_FILENO);
109 try ipc.appendMessage(
110 gpa,
111 &sock_write_buf,
112 .Resize,
113 std.mem.asBytes(&next_size),
114 );
115 }
116
117 // Handle stdin -> socket (Input)
118 const inp_flags = (lib_posix.POLL.IN | lib_posix.POLL.HUP | lib_posix.POLL.ERR | lib_posix.POLL.NVAL);
119 if (poll_fds.items[0].revents & inp_flags != 0) {
120 var buf: [4096]u8 = undefined;
121 const n_opt: ?usize = lib_posix.read(stdin_fd, &buf) catch |err| blk: {
122 if (err == error.WouldBlock) break :blk null;
123 return err;
124 };
125
126 if (n_opt) |n| {
127 if (n > 0) {
128 // Check for detach sequences (ctrl+\ as first byte or Kitty escape sequence)
129 if (!detach_key_disabled and util.isCtrlBackslash(buf[0..n])) {
130 std.log.info("detach key detected", .{});
131 try ipc.appendMessage(gpa, &sock_write_buf, .Detach, "");
132 } else {
133 try ipc.appendMessage(gpa, &sock_write_buf, .Input, buf[0..n]);
134 }
135 } else {
136 std.log.info("eof stdin", .{});
137 // EOF on stdin
138 return ClientResult{ .kind = .detach, .session_name = null };
139 }
140 }
141 }
142
143 // Handle socket read (incoming Output messages from daemon)
144 if (poll_fds.items[1].revents & lib_posix.POLL.IN != 0) {
145 const n = read_buf.read(client_sock_fd) catch |err| {
146 if (err == error.WouldBlock) continue;
147 if (err == error.ConnectionResetByPeer or err == error.BrokenPipe) {
148 return ClientResult{ .kind = .detach, .session_name = null };
149 }
150 std.log.err("daemon read err={s}", .{@errorName(err)});
151 return err;
152 };
153 if (n == 0) {
154 std.log.info("server closed connection", .{});
155 // Server closed connection
156 return ClientResult{ .kind = .detach, .session_name = null };
157 }
158
159 while (read_buf.next()) |msg| {
160 switch (msg.header.tag) {
161 .Output => {
162 if (msg.payload.len > 0) {
163 try stdout_buf.appendSlice(gpa, msg.payload);
164 }
165 },
166 .Resize => {
167 // daemon is asking for the client's window size usually in response
168 // to this client being set as leader.
169 const next_size = ipc.getTerminalSize(lib_posix.STDOUT_FILENO);
170 try ipc.appendMessage(
171 gpa,
172 &sock_write_buf,
173 .Resize,
174 std.mem.asBytes(&next_size),
175 );
176 },
177 .Switch => {
178 std.log.info("switch session", .{});
179 // Payload format: "session_name\ncwd" from the daemon
180 const newline_idx = std.mem.indexOfScalar(u8, msg.payload, '\n') orelse {
181 // No cwd provided (backward compat or old daemon)
182 return ClientResult{ .kind = .switch_session, .session_name = try gpa.dupe(u8, msg.payload) };
183 };
184 return ClientResult{
185 .kind = .switch_session,
186 .session_name = try gpa.dupe(u8, msg.payload[0..newline_idx]),
187 .cwd = if (newline_idx + 1 < msg.payload.len) try gpa.dupe(u8, msg.payload[newline_idx + 1 ..]) else null,
188 };
189 },
190 else => {},
191 }
192 }
193 }
194
195 // Handle socket write (flush buffered messages to daemon)
196 if (poll_fds.items[1].revents & lib_posix.POLL.OUT != 0) {
197 if (sock_write_buf.items.len > 0) {
198 const n = lib_posix.write(client_sock_fd, sock_write_buf.items) catch |err| blk: {
199 if (err == error.WouldBlock) break :blk 0;
200 if (err == error.ConnectionResetByPeer or err == error.BrokenPipe) {
201 std.log.info("connection reset or broken pipe", .{});
202 return ClientResult{ .kind = .detach, .session_name = null };
203 }
204 return err;
205 };
206 if (n > 0) {
207 try sock_write_buf.replaceRange(gpa, 0, n, &[_]u8{});
208 }
209 }
210 }
211
212 if (stdout_buf.items.len > 0) {
213 const n = lib_posix.write(lib_posix.STDOUT_FILENO, stdout_buf.items) catch |err| blk: {
214 if (err == error.WouldBlock) break :blk 0;
215 return err;
216 };
217 if (n > 0) {
218 try stdout_buf.replaceRange(gpa, 0, n, &[_]u8{});
219 }
220 }
221
222 if (poll_fds.items[1].revents & (lib_posix.POLL.HUP | lib_posix.POLL.ERR | lib_posix.POLL.NVAL) != 0) {
223 std.log.info("poll hup|err|nval", .{});
224 return ClientResult{ .kind = .detach, .session_name = null };
225 }
226 }
227}
228
229fn initTerminal(gpa: std.mem.Allocator, io: std.Io, size: ipc.Resize, cfg: *const Cfg) !ghostty_vt.Terminal {
230 return ghostty_vt.Terminal.init(io, gpa, .{
231 .cols = size.cols,
232 .rows = size.rows,
233 .max_scrollback_lines = cfg.max_scrollback_lines,
234 .max_scrollback_bytes = null, // Let the line limit control scrollback.
235 });
236}
237
238/// dameonLoop is what the daemon runs to send and receive ipc commands from its corresponding
239/// clients. It uses poll() as its non-blocking mechanism.
240fn daemonLoop(daemon: *Daemon, gpa: std.mem.Allocator, io: std.Io, server_sock_fd: lib_posix.socket_t, pty_fd: i32) !void {
241 std.log.info("daemon started session={s} pty_fd={d}", .{ daemon.session_name, pty_fd });
242
243 try signal.openSignalPipe();
244 signal.installWakeHandler(@backingInt(lib_posix.SIG.TERM));
245 var poll_fds = try std.ArrayList(lib_posix.pollfd).initCapacity(gpa, 8);
246 defer poll_fds.deinit(gpa);
247
248 const init_size = ipc.getTerminalSize(pty_fd);
249 var term = try initTerminal(gpa, io, init_size, daemon.cfg);
250 defer term.deinit(gpa);
251 var vt_stream = term.vtStream();
252 defer vt_stream.deinit();
253
254 // Carries the tail of the previous PTY read so the task-exit marker
255 // search below can see across a read() boundary. Sized to comfortably
256 // hold "ZMX_TASK_COMPLETED:" (19 bytes) plus a u8 exit code and CRLF.
257 var marker_carry: [32]u8 = undefined;
258 var marker_carry_len: usize = 0;
259
260 var had_terminal_client = daemon.hasTerminalClient();
261
262 daemon_loop: while (daemon.running) {
263 // If the program asked for focus reports (DECSET 1004), send focus-out
264 // when the last attached client leaves and focus-in when one returns,
265 // as a terminal would when its window loses/gains focus.
266 const has_terminal_client = daemon.hasTerminalClient();
267 if (has_terminal_client != had_terminal_client and term.modes.get(.focus_event)) {
268 daemon.queuePtyInput(gpa, if (has_terminal_client) "\x1b[I" else "\x1b[O");
269 }
270 had_terminal_client = has_terminal_client;
271
272 poll_fds.clearRetainingCapacity();
273
274 try poll_fds.append(gpa, .{
275 .fd = server_sock_fd,
276 .events = lib_posix.POLL.IN,
277 .revents = 0,
278 });
279
280 var pty_events: i16 = lib_posix.POLL.IN;
281 if (daemon.pty_write_buf.items.len > 0) {
282 pty_events |= lib_posix.POLL.OUT;
283 }
284 try poll_fds.append(gpa, .{
285 .fd = pty_fd,
286 .events = pty_events,
287 .revents = 0,
288 });
289
290 try poll_fds.append(gpa, .{ .fd = signal.sig_pipe[0], .events = lib_posix.POLL.IN, .revents = 0 });
291
292 for (daemon.clients.items) |client| {
293 var events: i16 = lib_posix.POLL.IN;
294 if (client.has_pending_output) {
295 events |= lib_posix.POLL.OUT;
296 }
297 try poll_fds.append(gpa, .{
298 .fd = client.socket_fd,
299 .events = events,
300 .revents = 0,
301 });
302 }
303
304 _ = try lib_posix.poll(poll_fds.items, -1);
305
306 if (poll_fds.items[2].revents & lib_posix.POLL.IN != 0) {
307 signal.drainSignalPipe();
308 std.log.info(
309 "SIGTERM received, shutting down gracefully session={s}",
310 .{daemon.session_name},
311 );
312 break :daemon_loop;
313 }
314
315 if (poll_fds.items[0].revents & (lib_posix.POLL.ERR | lib_posix.POLL.HUP | lib_posix.POLL.NVAL) != 0) {
316 std.log.err("server socket error revents={d}", .{poll_fds.items[0].revents});
317 break :daemon_loop;
318 } else if (poll_fds.items[0].revents & lib_posix.POLL.IN != 0) {
319 const client_fd = try lib_posix.accept(
320 server_sock_fd,
321 null,
322 null,
323 lib_posix.SOCK.NONBLOCK | lib_posix.SOCK.CLOEXEC,
324 );
325 const client = try gpa.create(Client);
326 client.* = Client{
327 .alloc = gpa,
328 .socket_fd = client_fd,
329 .read_buf = try ipc.SocketBuffer.init(gpa),
330 .write_buf = undefined,
331 };
332 // 64KB initial capacity lets ~15 broadcast cycles (N_TTY_BUF_SIZE reads
333 // * header) accumulate before the first ArrayList growth. The write
334 // buffer is userspace-only: it drains via POLLOUT to the client socket,
335 // which has no corresponding kernel-imposed per-write limit.
336 client.write_buf = try std.ArrayList(u8).initCapacity(client.alloc, 65536);
337 try daemon.clients.append(gpa, client);
338 std.log.info(
339 "client connected fd={d} total={d}",
340 .{ client_fd, daemon.clients.items.len },
341 );
342 }
343
344 const inp_flags = lib_posix.POLL.IN | lib_posix.POLL.HUP | lib_posix.POLL.ERR | lib_posix.POLL.NVAL;
345 if (poll_fds.items[1].revents & inp_flags != 0) {
346 // Read from PTY. Buffer is sized to N_TTY_BUF_SIZE (4096): the hard
347 // kernel limit for the N_TTY line discipline. A larger buffer doesn't
348 // help: each read() from a PTY master returns at most 4096 bytes
349 // regardless of the userspace buffer size.
350 var buf: [4096]u8 = undefined;
351 const n_opt: ?usize = lib_posix.read(pty_fd, &buf) catch |err| blk: {
352 if (err == error.WouldBlock) break :blk null;
353 break :blk 0;
354 };
355
356 if (n_opt) |n| {
357 if (n == 0) {
358 // EOF: Shell exited
359 std.log.info("shell exited pty_fd={d}", .{pty_fd});
360 // Let the rest of this poll iteration complete so client
361 // write buffers are flushed via the normal POLLOUT path.
362 // On the next iteration, daemon.running will be false.
363 daemon.running = false;
364 } else {
365 // Feed PTY output to terminal emulator for state tracking
366 vt_stream.nextSlice(buf[0..n]);
367 daemon.setPwd(&term);
368 daemon.has_pty_output = true;
369
370 // When no real terminal client has attached yet, respond to
371 // terminal queries (e.g. DA1/DA2) on behalf of the terminal.
372 // This prevents fish from waiting 10s for unanswered queries.
373 // Only clients that sent .Init (a real zmx attach) count,
374 // not a `zmx run` tail-only client.
375 if (!daemon.hasTerminalClient() and
376 daemon.pty_write_buf.items.len < Daemon.PTY_WRITE_BUF_MAX)
377 {
378 util.respondToDeviceAttributes(gpa, &daemon.pty_write_buf, buf[0..n]);
379 }
380
381 // In run mode, scan output for exit code marker. The marker
382 // can straddle two PTY reads (more likely under a throttled
383 // scheduler, e.g. containers), so prepend the tail carried
384 // over from the previous read before searching.
385 if (daemon.is_task_mode and daemon.task_exit_code == null) {
386 var scan_buf: [marker_carry.len + buf.len]u8 = undefined;
387 @memcpy(scan_buf[0..marker_carry_len], marker_carry[0..marker_carry_len]);
388 @memcpy(scan_buf[marker_carry_len..][0..n], buf[0..n]);
389 const scan_len = marker_carry_len + n;
390
391 if (try util.findTaskExitMarker(scan_buf[0..scan_len], daemon.task_id)) |exit_code| {
392 daemon.task_exit_code = exit_code;
393 daemon.task_ended_at = @intCast(std.Io.Timestamp.now(io, .real).toSeconds());
394
395 std.log.info("task completed exit_code={d}", .{exit_code});
396
397 // Notify connected clients
398 for (daemon.clients.items) |c| {
399 ipc.appendMessage(gpa, &c.write_buf, .TaskComplete, &[_]u8{exit_code}) catch {};
400 c.has_pending_output = true;
401 }
402 }
403
404 marker_carry_len = @min(marker_carry.len, scan_len);
405 @memcpy(
406 marker_carry[0..marker_carry_len],
407 scan_buf[scan_len - marker_carry_len .. scan_len],
408 );
409 }
410
411 // Broadcast data to all clients.
412 // Rewrite OSC 133;A to include redraw=0 so the outer terminal
413 // does not clear prompt lines on resize (issue #111).
414 const broadcast_data = util.rewritePromptRedraw(gpa, buf[0..n]) orelse buf[0..n];
415 defer if (broadcast_data.ptr != buf[0..n].ptr) gpa.free(broadcast_data);
416 for (daemon.clients.items) |client| {
417 ipc.appendMessage(gpa, &client.write_buf, .Output, broadcast_data) catch |err| {
418 std.log.warn(
419 "failed to buffer output for client err={s}",
420 .{@errorName(err)},
421 );
422 continue;
423 };
424 client.has_pending_output = true;
425 }
426 }
427 }
428 }
429
430 if (poll_fds.items[1].revents & lib_posix.POLL.OUT != 0) {
431 while (daemon.pty_write_buf.items.len > 0) {
432 const n = lib_posix.write(pty_fd, daemon.pty_write_buf.items) catch |err| {
433 if (err != error.WouldBlock) {
434 std.log.warn("pty write failed: {s}", .{@errorName(err)});
435 daemon.pty_write_buf.clearRetainingCapacity();
436 }
437 break;
438 };
439 if (n == 0) break;
440 daemon.pty_write_buf.replaceRange(gpa, 0, n, &[_]u8{}) catch unreachable;
441 }
442 }
443
444 var i: usize = daemon.clients.items.len;
445 // Only iterate over clients that were present when poll_fds was constructed
446 // poll_fds contains [server, pty, sig_pipe, client0, client1, ...]
447 // So number of clients in poll_fds is poll_fds.items.len - 3
448 const num_polled_clients = poll_fds.items.len - 3;
449 if (i > num_polled_clients) {
450 // If we have more clients than polled (i.e. we just accepted one), start from the
451 // polled ones
452 i = num_polled_clients;
453 }
454
455 clients_loop: while (i > 0) {
456 i -= 1;
457 const client = daemon.clients.items[i];
458 const revents = poll_fds.items[i + 3].revents;
459
460 if (revents & lib_posix.POLL.IN != 0) {
461 const n = client.read_buf.read(client.socket_fd) catch |err| {
462 if (err == error.WouldBlock) continue;
463 std.log.debug(
464 "client read err={s} fd={d}",
465 .{ @errorName(err), client.socket_fd },
466 );
467 const last = daemon.closeClient(gpa, client, i, false);
468 if (last) break :daemon_loop;
469 continue;
470 };
471
472 if (n == 0) {
473 // Client closed connection
474 const last = daemon.closeClient(gpa, client, i, false);
475 if (last) break :daemon_loop;
476 continue;
477 }
478
479 while (client.read_buf.next()) |msg| {
480 switch (msg.header.tag) {
481 .Input => try daemon.handleInput(gpa, client, msg.payload),
482 .Send => daemon.handleSend(gpa, msg.payload),
483 .Output => try daemon.handleOutput(gpa, msg.payload, &term, &vt_stream),
484 .Init => try daemon.handleInit(gpa, client, pty_fd, &term, msg.payload),
485 .Switch => try daemon.handleSwitch(gpa, msg.payload),
486 .Resize => try daemon.handleResize(gpa, client, pty_fd, &term, msg.payload),
487 .Detach => {
488 daemon.handleDetach(gpa, client, i);
489 break :clients_loop;
490 },
491 .DetachAll => {
492 daemon.handleDetachAll(gpa);
493 break :clients_loop;
494 },
495 .Kill => {
496 break :daemon_loop;
497 },
498 .Info => try daemon.handleInfo(gpa, client, &term),
499 .LabelGet => try daemon.handleLabelGet(gpa, client),
500 .LabelSet => try daemon.handleLabelSet(gpa, client, msg.payload),
501 .LabelClear => try daemon.handleLabelClear(gpa, client),
502 .EnvGet => try daemon.handleEnvGet(gpa, client),
503 .EnvSet => try daemon.handleEnvSet(gpa, client, msg.payload),
504 .History => try daemon.handleHistory(gpa, client, &term, msg.payload),
505 .Capture => try daemon.handleCapture(gpa, client, &term, msg.payload),
506 .Run => try daemon.handleRun(gpa, io, client, msg.payload),
507 .Ack, .TaskComplete, .LabelData, .EnvData, .Unsupported => {},
508 .Write => try daemon.handleWrite(gpa, client, msg.payload),
509 _ => {
510 std.log.warn(
511 "ignoring unknown IPC tag={d}",
512 .{@backingInt(msg.header.tag)},
513 );
514 ipc.appendMessage(gpa, &client.write_buf, .Unsupported, "") catch {};
515 client.has_pending_output = true;
516 },
517 }
518 }
519 }
520
521 if (revents & lib_posix.POLL.OUT != 0) {
522 // Flush pending output buffers
523 const n = lib_posix.write(client.socket_fd, client.write_buf.items) catch |err| blk: {
524 if (err == error.WouldBlock) break :blk 0;
525 // Error on write, close client
526 const last = daemon.closeClient(gpa, client, i, false);
527 if (last) break :daemon_loop;
528 continue;
529 };
530
531 if (n > 0) {
532 client.write_buf.replaceRange(gpa, 0, n, &[_]u8{}) catch unreachable;
533 }
534
535 if (client.write_buf.items.len == 0) {
536 client.has_pending_output = false;
537 }
538 }
539
540 if (revents & (lib_posix.POLL.HUP | lib_posix.POLL.ERR | lib_posix.POLL.NVAL) != 0) {
541 // A client that writes and exits (`zmx send`, `zmx run -d`) is
542 // reported HUP with part of its message still unread: a pass
543 // reads at most 4 KiB. Closing now drops the rest. Leave it to
544 // the next pass, which reads on until EOF closes the client.
545 if (revents & lib_posix.POLL.IN != 0 and
546 revents & (lib_posix.POLL.ERR | lib_posix.POLL.NVAL) == 0) continue;
547 const last = daemon.closeClient(gpa, client, i, false);
548 if (last) break :daemon_loop;
549 }
550 }
551 }
552}
553
554const ClientResult = struct {
555 kind: enum {
556 detach,
557 switch_session,
558 },
559 session_name: ?[]const u8,
560 cwd: ?[]const u8 = null,
561};
562
563/// Client represents each terminal that has connected to a session.
564///
565/// Multiple Clients can connect to a single session.
566pub const Client = struct {
567 alloc: std.mem.Allocator,
568 socket_fd: i32,
569 has_pending_output: bool = false,
570 is_terminal: bool = false, // sent .Init (a `zmx attach`), not a run/send/tail client
571 read_buf: ipc.SocketBuffer,
572 write_buf: std.ArrayList(u8),
573 env_str: ?[]u8 = null,
574
575 pub fn deinit(self: *Client, gpa: std.mem.Allocator) void {
576 lib_posix.close(self.socket_fd);
577 self.read_buf.deinit();
578 self.write_buf.deinit(self.alloc);
579 if (self.env_str) |s| gpa.free(s);
580 }
581
582 fn setEnv(self: *Client, gpa: std.mem.Allocator, env_str: []const u8) !void {
583 if (self.env_str) |s| gpa.free(s);
584 self.env_str = if (env_str.len > 0) try gpa.dupe(u8, env_str) else null;
585 }
586};
587
588/// Daemon is responsible for managing a zmx session.
589///
590/// It holds all the state for a running session. Instead of a single daemon for all sessions, we
591/// create a daemon for every session. This has some benefits. The ipc communication between
592/// session clients and the daemon doesn't need to be tagged with the session name. If a daemon
593/// crashes for one session won't crash all the other sessions.
594///
595/// Conceptually it's also much simpler to reason about.
596pub const Daemon = struct {
597 cfg: *Cfg,
598 session_name: []const u8,
599 socket_path: []const u8,
600 // === opt ===
601 pty_write_buf: std.ArrayList(u8) = .empty,
602 clients: std.ArrayList(*Client) = .empty,
603 labels: std.StringHashMapUnmanaged([]u8) = .empty,
604 // This control which client is the leader. The leader controls terminal state and
605 // cols/rows of session.
606 leader_client_fd: ?i32 = null,
607 running: bool = true,
608 pid: i32 = undefined,
609 command: ?[]const []const u8 = null,
610 /// The session's working directory in OSC 7 form, `file://<host><path>`.
611 /// Kept as a URI rather than a path so `zmx list` shows the host, which is
612 /// what tells you a session is inside SSH. Points into `cwd_buf` once set,
613 /// so a Daemon must not be copied by value after that.
614 cwd: []const u8 = "",
615 /// The same directory as a path that can be opened: percent-decoding
616 /// applied, scheme and host stripped. Empty when the cwd is on another
617 /// host, since then it names no directory here and nothing should chdir
618 /// into it. Points into `cwd_path_buf`.
619 cwd_path: []const u8 = "",
620 cwd_buf: [std.fs.max_path_bytes]u8 = undefined,
621 cwd_path_buf: [std.fs.max_path_bytes]u8 = undefined,
622 has_pty_output: bool = false,
623 created_at: u64, // unix timestamp (ns)
624 is_task_mode: bool = false, // flag for when session is run as a task
625 task_id: [4]u8 = undefined,
626 task_exit_code: ?u8 = null, // null = running or n/a, set when task completes
627 task_ended_at: ?u64 = null, // timestamp when task exited
628 pty_fd: i32 = -1, // set by daemonLoop so handleRun can probe the foreground process
629 shell: []const u8 = "/bin/sh",
630 quiet: bool = false,
631
632 /// Create a Daemon. Caller is responsible for freeing all variables passed
633 /// into the init fn.
634 pub fn init(io: std.Io, cfg: *Cfg, sesh_name: []const u8, socket_path: []const u8) Daemon {
635 return .{
636 .cfg = cfg,
637 .session_name = sesh_name,
638 .socket_path = socket_path,
639 .created_at = @intCast(std.Io.Timestamp.now(io, .real).toSeconds()),
640 };
641 }
642
643 pub fn deinit(self: *Daemon, gpa: std.mem.Allocator) void {
644 self.clients.deinit(gpa);
645 var it = self.labels.iterator();
646 while (it.next()) |entry| {
647 gpa.free(entry.key_ptr.*);
648 gpa.free(entry.value_ptr.*);
649 }
650 self.labels.deinit(gpa);
651 self.pty_write_buf.deinit(gpa);
652 gpa.free(self.socket_path);
653 }
654
655 pub fn shutdown(self: *Daemon, gpa: std.mem.Allocator) void {
656 std.log.info("shutting down daemon session={s}", .{self.session_name});
657 self.running = false;
658
659 for (self.clients.items) |client| {
660 client.deinit(gpa);
661 gpa.destroy(client);
662 }
663 self.clients.clearRetainingCapacity();
664 }
665
666 /// Resize the daemon's terminal with prompt_redraw disabled. On resize the
667 /// terminal would clear prompt lines expecting the shell to redraw them,
668 /// but the shell's redraw goes to the PTY (forwarded to clients), not to
669 /// this terminal, so the clearing only corrupts our snapshot state.
670 fn resizeTerm(gpa: std.mem.Allocator, term: *ghostty_vt.Terminal, cols: u16, rows: u16) !void {
671 const saved = term.flags.shell_redraws_prompt;
672 term.flags.shell_redraws_prompt = .false;
673 defer term.flags.shell_redraws_prompt = saved;
674 try term.resize(gpa, .{ .cols = cols, .rows = rows });
675 }
676
677 /// True while a client that sent .Init (a real `zmx attach`) is connected.
678 fn hasTerminalClient(self: *const Daemon) bool {
679 for (self.clients.items) |c| {
680 if (c.is_terminal) return true;
681 }
682 return false;
683 }
684
685 pub fn closeClient(self: *Daemon, gpa: std.mem.Allocator, client: *Client, i: usize, shutdown_on_last: bool) bool {
686 const fd = client.socket_fd;
687 // leader is disconnected, remove ref and let another client claim leader on input
688 if (self.leader_client_fd == client.socket_fd) {
689 std.log.info(
690 "unsetting leader session={s} fd={d}",
691 .{ self.session_name, client.socket_fd },
692 );
693 self.leader_client_fd = null;
694 }
695 client.deinit(gpa);
696 gpa.destroy(client);
697 _ = self.clients.orderedRemove(i);
698 std.log.info("client disconnected fd={d} remaining={d}", .{ fd, self.clients.items.len });
699 if (shutdown_on_last and self.clients.items.len == 0) {
700 self.shutdown(gpa);
701 return true;
702 }
703 return false;
704 }
705
706 /// ensureSession will either create or re-use the daemon used for a session.
707 /// It will spin up a unix socket, double-fork the process (so it survives
708 /// the terminal dying), and automatically attach the client to the ipc unix
709 /// socket.
710 ///
711 /// The return bool value indicates if the current process is the daemon
712 /// or the client since they have different behaviors post-fork.
713 ///
714 /// E.g. If it's the client process then we need to connect to the unix socket
715 /// and run the clientLoop. If it's the daemon then we need to bail since
716 /// the daemonLoop is created inside this fn and when it returns that means
717 /// the daemon stopped and needs to exit.
718 pub fn ensureSession(self: *Daemon, io: std.Io) !bool {
719 const sesh_name = self.session_name;
720 std.log.info("ensure session session={s}", .{sesh_name});
721 var dir = try std.Io.Dir.openDirAbsolute(io, self.cfg.socket_dir, .{});
722 defer dir.close(io);
723
724 const exists = try socket.sessionExists(io, dir, sesh_name);
725 // if daemon is gone then we flip this to true
726 var should_create = !exists;
727
728 if (exists) {
729 if (ipc.connectSession(self.socket_path)) |fd| {
730 lib_posix.close(fd);
731 if (self.command != null) {
732 std.log.warn(
733 "session already exists, ignoring command session={s}",
734 .{sesh_name},
735 );
736 }
737 } else |err| switch (err) {
738 // Daemon is definitively gone: safe to replace.
739 error.ConnectionRefused => {
740 socket.cleanupStaleSocket(io, dir, sesh_name);
741 should_create = true;
742 },
743 // Connect failed for an unusual reason. The check is only to
744 // decide create-vs-attach; the socket file exists, so proceed
745 // to attach rather than fail or orphan.
746 else => {
747 std.log.warn(
748 "connect failed ({s}), proceeding to attach session={s}",
749 .{ @errorName(err), sesh_name },
750 );
751 },
752 }
753 }
754
755 if (!should_create) {
756 return false;
757 }
758
759 return self.run(io, dir, sesh_name);
760 }
761
762 fn run(self: *Daemon, io: std.Io, dir: std.Io.Dir, sesh_name: []const u8) !bool {
763 std.log.info("creating session={s}", .{sesh_name});
764 const server_sock_fd: lib_posix.socket_t = try socket.createSocket(self.socket_path);
765 const log_fd = if (log.log_system.file) |f| f.handle else -1;
766
767 var keep_fds_open = [_]i32{ server_sock_fd, dir.handle, log_fd };
768 const cmd = try daemonize.createCmdZ(self.shell, self.is_task_mode, self.command);
769
770 // `cwd_path` is the decoded path, and is empty when the cwd is on
771 // another host: OSC 7 crosses SSH boundaries, so a session that ssh'd
772 // elsewhere reports a directory that does not exist on this machine.
773 std.log.info("checking pwd={s} path={s}", .{ self.cwd, self.cwd_path });
774 if (self.cwd_path.len > 0) {
775 const pwd_dir = std.Io.Dir.openDirAbsolute(io, self.cwd_path, .{}) catch |err| blk: {
776 std.log.warn("failed to open dir={s} err={s}", .{ self.cwd_path, @errorName(err) });
777 break :blk null;
778 };
779 if (pwd_dir) |pdir| {
780 defer std.Io.Dir.close(pdir, io);
781 std.log.info("set directory dir={s}", .{self.cwd_path});
782 try std.process.setCurrentDir(io, pdir);
783 }
784 }
785
786 const pty_info = daemonize.daemonize(
787 sesh_name,
788 cmd,
789 &keep_fds_open,
790 ) catch |err| {
791 switch (err) {
792 error.IsClientProc => {
793 // send a msg to the client that the session was created.
794 if (!self.quiet) {
795 var w_buf: [2048]u8 = undefined;
796 var w = std.Io.File.stdout().writer(io, &w_buf);
797 try w.interface.print("session \"{s}\" created\n", .{sesh_name});
798 try w.interface.flush();
799 }
800 lib_posix.close(server_sock_fd);
801 return false;
802 },
803 else => {
804 lib_posix.close(server_sock_fd);
805 dir.deleteFile(io, self.session_name) catch {};
806 return err;
807 },
808 }
809 };
810 // =======
811 // WARNING: cannot use upstream allocator or io after this point since
812 // we forked the process and there's a risk of a mutex (e.g. thread-safe
813 // allocator) being locked by a thread prior to fork which can cause a
814 // deadlock.
815 // =======
816
817 self.pid = pty_info.pid;
818
819 var threaded: std.Io.Threaded = .init_single_threaded;
820 defer threaded.deinit();
821 const new_io = threaded.io();
822
823 { // re-initialize logs with the session name as the filename
824 log.log_system.deinit();
825 var log_buf: [4096]u8 = undefined;
826 const session_log_name = try std.fmt.bufPrint(
827 &log_buf,
828 "{s}.log",
829 .{sesh_name},
830 );
831 var fba_buf: [4096]u8 = undefined;
832 var fba = std.heap.FixedBufferAllocator.init(&fba_buf);
833 const session_log_path = try std.fs.path.join(
834 fba.allocator(),
835 &.{ self.cfg.log_dir, session_log_name },
836 );
837 const log_mode = std.Io.File.Permissions.fromMode(@intCast(self.cfg.log_mode));
838 log.log_system.init(new_io, session_log_path, log_mode) catch {};
839 }
840
841 const gpa: std.mem.Allocator = blk: {
842 if (builtin.mode == .debug) {
843 const GPA = std.heap.DebugAllocator(.{});
844 const Static = struct {
845 var gpa: GPA = .{};
846 };
847 break :blk Static.gpa.allocator();
848 }
849 break :blk std.heap.c_allocator;
850 };
851
852 defer {
853 // Close and unlink the listen socket BEFORE handleKill()'s
854 // 500ms SIGHUP->SIGKILL grace sleep. Otherwise a `zmx run`
855 // for the same name issued in that window will hang waiting
856 // for a connect.
857 lib_posix.close(server_sock_fd);
858 std.log.info("deleting socket file session={s}", .{sesh_name});
859 dir.deleteFile(new_io, sesh_name) catch |err| {
860 std.log.warn("failed to delete socket file err={s}", .{@errorName(err)});
861 };
862 self.handleKill(gpa, new_io);
863 self.deinit(gpa);
864 lib_posix.close(pty_info.master_fd);
865 _ = lib_posix.waitpid(self.pid, 0);
866 }
867
868 try daemonLoop(self, gpa, new_io, server_sock_fd, pty_info.master_fd);
869 std.log.info("daemon loop shutdown", .{});
870 return true;
871 }
872
873 fn setLeader(self: *Daemon, gpa: std.mem.Allocator, client: *Client) !void {
874 std.log.info("setting new leader client_fd={d}", .{client.socket_fd});
875 self.leader_client_fd = client.socket_fd;
876 // Send a resize message to the client so it can send us back their window size
877 // so we can resize the pty and ghostty state.
878 try ipc.appendMessage(gpa, &client.write_buf, .Resize, "");
879 client.has_pending_output = true;
880 }
881
882 fn getLeaderClient(self: *Daemon) ?*Client {
883 for (self.clients.items) |client| {
884 if (self.leader_client_fd == client.socket_fd) {
885 return client;
886 }
887 }
888 return null;
889 }
890
891 const PTY_WRITE_BUF_MAX = 256 * 1024;
892
893 /// Queue bytes for the PTY's stdin. Flushed by daemonLoop on POLLOUT.
894 /// Drops the payload if the buffer is over cap -- same failure mode as
895 /// the old direct-write ptyWrite (drop on EAGAIN), just at a 64x higher
896 /// threshold. Capping avoids OOM when the shell stops reading; dropping
897 /// new (not old) bytes avoids tearing a partially-accepted sequence.
898 fn queuePtyInput(self: *Daemon, gpa: std.mem.Allocator, data: []const u8) void {
899 if (data.len == 0) return;
900 if (self.pty_write_buf.items.len + data.len > PTY_WRITE_BUF_MAX) {
901 std.log.warn(
902 "pty input dropped {d} bytes (buffer full, shell not reading)",
903 .{data.len},
904 );
905 return;
906 }
907
908 // NOTE: for local dev only
909 // std.log.debug("buffering pty input data={x}", .{data});
910
911 self.pty_write_buf.appendSlice(gpa, data) catch |err| {
912 std.log.warn(
913 "pty input dropped {d} bytes: {s}",
914 .{ data.len, @errorName(err) },
915 );
916 };
917 }
918
919 pub fn handleInput(self: *Daemon, gpa: std.mem.Allocator, client: *Client, payload: []const u8) !void {
920 // NOTE: for local dev only
921 // std.log.debug("buffering pty input data={x}", .{payload});
922
923 // client is leader, send entire payload (ansi escape codes + text)
924 if (self.leader_client_fd == client.socket_fd) {
925 self.queuePtyInput(gpa, payload);
926 return;
927 }
928
929 // check if leader needs to be updated by detecting any user input
930 if (util.isUserInput(payload)) {
931 try self.setLeader(gpa, client);
932 self.queuePtyInput(gpa, payload);
933 }
934 }
935
936 /// Queue input from `zmx send` without changing interactive client leadership.
937 pub fn handleSend(self: *Daemon, gpa: std.mem.Allocator, payload: []const u8) void {
938 self.queuePtyInput(gpa, payload);
939 }
940
941 pub fn handleSwitch(self: *Daemon, gpa: std.mem.Allocator, session_name: []const u8) !void {
942 for (self.clients.items) |client| {
943 if (self.leader_client_fd == client.socket_fd) {
944 // Include the daemon's current cwd so the new session can start
945 // in the right directory. A remote cwd is left out: it names no
946 // directory here, so the new session is better off with the
947 // attaching client's own cwd than with a path it cannot enter.
948 if (self.cwd.len > 0 and self.cwd_path.len > 0) {
949 var payload = gpa.alloc(u8, session_name.len + 1 + self.cwd.len) catch return;
950 defer gpa.free(payload);
951 @memcpy(payload[0..session_name.len], session_name);
952 payload[session_name.len] = '\n';
953 @memcpy(payload[session_name.len + 1 ..], self.cwd);
954 ipc.appendMessage(gpa, &client.write_buf, .Switch, payload) catch |err| {
955 std.log.warn(
956 "failed to buffer terminal state for client err={s}",
957 .{@errorName(err)},
958 );
959 };
960 } else {
961 ipc.appendMessage(gpa, &client.write_buf, .Switch, session_name) catch |err| {
962 std.log.warn(
963 "failed to buffer terminal state for client err={s}",
964 .{@errorName(err)},
965 );
966 };
967 }
968 client.has_pending_output = true;
969 return;
970 }
971 }
972 }
973
974 pub fn handleInit(
975 self: *Daemon,
976 gpa: std.mem.Allocator,
977 client: *Client,
978 pty_fd: i32,
979 term: *ghostty_vt.Terminal,
980 payload: []const u8,
981 ) !void {
982 if (payload.len != @sizeOf(ipc.Resize)) return;
983
984 if (!client.is_terminal) {
985 client.write_buf.clearRetainingCapacity();
986 client.has_pending_output = false;
987 }
988 client.is_terminal = true;
989
990 if (self.leader_client_fd == null) {
991 try self.setLeader(gpa, client);
992 }
993 const is_leader = self.leader_client_fd == client.socket_fd;
994 const resize = std.mem.bytesToValue(ipc.Resize, payload);
995
996 // Resize our terminal (not yet the PTY) to the leader's size before
997 // serializing, so the snapshot is laid out for the width the client
998 // will render it at instead of being wrapped a second time on arrival.
999 // Cursor position stays consistent because it is serialized from the
1000 // same, already-resized terminal; the PTY is resized below, so the
1001 // shell's own SIGWINCH redraw still arrives after the snapshot.
1002 if (is_leader) {
1003 try resizeTerm(gpa, term, resize.cols, resize.rows);
1004 }
1005
1006 if (self.has_pty_output) {
1007 const cursor = &term.screens.active.cursor;
1008 std.log.debug(
1009 "cursor before serialize: x={d} y={d} pending_wrap={}",
1010 .{ cursor.x, cursor.y, cursor.pending_wrap },
1011 );
1012 if (util.serializeTerminalState(gpa, term)) |term_output| {
1013 std.log.debug("serialize terminal state", .{});
1014 // Rewrite OSC 133;A to include redraw=0 so the outer terminal
1015 // does not clear prompt lines on resize (issue #111).
1016 const restore_data = util.rewritePromptRedraw(gpa, term_output) orelse term_output;
1017 defer gpa.free(term_output);
1018 defer if (restore_data.ptr != term_output.ptr) gpa.free(restore_data);
1019 ipc.appendMessage(gpa, &client.write_buf, .Output, restore_data) catch |err| {
1020 std.log.warn(
1021 "failed to buffer terminal state for client err={s}",
1022 .{@errorName(err)},
1023 );
1024 };
1025 client.has_pending_output = true;
1026 }
1027 }
1028
1029 // only resize if leader
1030 if (is_leader) {
1031 var ws = resize.winsize();
1032 _ = cross.ioctl(pty_fd, cross.TIOCSWINSZ, &ws);
1033
1034 // On re-attach, deliver SIGWINCH to the foreground process group so
1035 // incremental renderers (Ink, Claude Code, etc.) know to repaint.
1036 // If the size changed, TIOCSWINSZ above already sent SIGWINCH; if the size
1037 // was unchanged, the kernel suppressed it, so signal the pgrp explicitly.
1038 if (self.has_pty_output) {
1039 var pgrp: lib_posix.pid_t = 0;
1040 if (cross.ioctl(pty_fd, cross.TIOCGPGRP, &pgrp) == 0 and pgrp > 0) {
1041 lib_posix.kill(-pgrp, .WINCH) catch {};
1042 }
1043 }
1044
1045 std.log.debug("init resize rows={d} cols={d}", .{ resize.rows, resize.cols });
1046 }
1047 }
1048
1049 pub fn handleResize(
1050 self: *Daemon,
1051 gpa: std.mem.Allocator,
1052 client: *Client,
1053 pty_fd: i32,
1054 term: *ghostty_vt.Terminal,
1055 payload: []const u8,
1056 ) !void {
1057 if (payload.len != @sizeOf(ipc.Resize)) return;
1058 if (self.leader_client_fd == null) {
1059 try self.setLeader(gpa, client);
1060 }
1061 // only leader can resize
1062 if (self.leader_client_fd != client.socket_fd) return;
1063
1064 const resize = std.mem.bytesToValue(ipc.Resize, payload);
1065 var ws = resize.winsize();
1066 _ = cross.ioctl(pty_fd, cross.TIOCSWINSZ, &ws);
1067 try resizeTerm(gpa, term, resize.cols, resize.rows);
1068 std.log.debug("resize rows={d} cols={d}", .{ resize.rows, resize.cols });
1069 }
1070
1071 pub fn handleDetach(self: *Daemon, gpa: std.mem.Allocator, client: *Client, i: usize) void {
1072 std.log.info("client detach session={s} fd={d}", .{ self.session_name, client.socket_fd });
1073 _ = self.closeClient(gpa, client, i, false);
1074 }
1075
1076 pub fn handleDetachAll(self: *Daemon, gpa: std.mem.Allocator) void {
1077 std.log.info("detach all clients={d}", .{self.clients.items.len});
1078 // Go through closeClient so the leader is cleared like any other detach.
1079 while (self.clients.items.len > 0) {
1080 const last = self.clients.items.len - 1;
1081 _ = self.closeClient(gpa, self.clients.items[last], last, false);
1082 }
1083 }
1084
1085 pub fn handleKill(self: *Daemon, gpa: std.mem.Allocator, io: std.Io) void {
1086 std.log.info("kill received session={s}", .{self.session_name});
1087 self.shutdown(gpa);
1088 // gracefully shutdown shell processes, shells tend to ignore SIGTERM so we send SIGHUP
1089 // instead
1090 // https://www.gnu.org/software/bash/manual/html_node/Signals.html
1091 // negative pid means kill process and children
1092 std.log.info("sending SIGHUP session={s} pid={d}", .{ self.session_name, self.pid });
1093 lib_posix.kill(-self.pid, lib_posix.SIG.HUP) catch |err| {
1094 std.log.warn("failed to send SIGHUP to pty child err={s}", .{@errorName(err)});
1095 };
1096 std.Io.sleep(io, std.Io.Duration.fromMilliseconds(500), .real) catch unreachable;
1097 lib_posix.kill(-self.pid, lib_posix.SIG.KILL) catch |err| {
1098 std.log.warn("failed to send SIGKILL to pty child err={s}", .{@errorName(err)});
1099 };
1100 }
1101
1102 pub fn handleInfo(self: *Daemon, gpa: std.mem.Allocator, client: *Client, term: *ghostty_vt.Terminal) !void {
1103 self.setPwd(term);
1104
1105 // zeroes() so asBytes() doesn't ship struct padding + unused cmd/cwd
1106 // tail bytes (daemon stack contents) to clients.
1107 var info = std.mem.zeroes(ipc.Info);
1108 info.protocol_version = ipc.PROTOCOL_VERSION;
1109 info.clients_len = self.clients.items.len - 1;
1110 info.pid = self.pid;
1111 info.created_at = self.created_at;
1112 info.task_ended_at = self.task_ended_at orelse 0;
1113 info.task_exit_code = self.task_exit_code orelse 0;
1114
1115 // Build command string from args, re-quoting args that contain
1116 // shell-special characters so the displayed command is copy-pasteable.
1117 const cur_cmd = self.command;
1118 if (cur_cmd) |args| {
1119 for (args, 0..) |arg, i| {
1120 const quoted = if (util.shellNeedsQuoting(arg))
1121 util.shellQuote(gpa, arg) catch null
1122 else
1123 null;
1124 defer if (quoted) |q| gpa.free(q);
1125 const src = quoted orelse arg;
1126
1127 const need = src.len + @as(usize, if (i > 0) 1 else 0);
1128 if (info.cmd_len + need > ipc.MAX_CMD_LEN) {
1129 const ellipsis = "...";
1130 if (info.cmd_len + ellipsis.len <= ipc.MAX_CMD_LEN) {
1131 @memcpy(info.cmd[info.cmd_len..][0..ellipsis.len], ellipsis);
1132 info.cmd_len += ellipsis.len;
1133 }
1134 break;
1135 }
1136
1137 if (i > 0) {
1138 info.cmd[info.cmd_len] = ' ';
1139 info.cmd_len += 1;
1140 }
1141 @memcpy(info.cmd[info.cmd_len..][0..src.len], src);
1142 info.cmd_len += @intCast(src.len);
1143 }
1144 }
1145
1146 info.cwd_len = @intCast(@min(self.cwd.len, ipc.MAX_CWD_LEN));
1147 @memcpy(info.cwd[0..info.cwd_len], self.cwd[0..info.cwd_len]);
1148
1149 try ipc.appendMessage(gpa, &client.write_buf, .Info, std.mem.asBytes(&info));
1150 client.has_pending_output = true;
1151 }
1152
1153 pub fn handleHistory(
1154 self: *Daemon,
1155 gpa: std.mem.Allocator,
1156 client: *Client,
1157 term: *ghostty_vt.Terminal,
1158 payload: []const u8,
1159 ) !void {
1160 self.setPwd(term);
1161 const format: util.HistoryFormat = if (payload.len > 0)
1162 @fromBackingInt(@intCast(payload[0]))
1163 else
1164 .plain;
1165 if (util.serializeTerminal(gpa, term, format)) |output| {
1166 defer gpa.free(output);
1167 try ipc.appendMessage(gpa, &client.write_buf, .History, output);
1168 client.has_pending_output = true;
1169 } else {
1170 try ipc.appendMessage(gpa, &client.write_buf, .History, "");
1171 client.has_pending_output = true;
1172 }
1173 }
1174
1175 pub fn handleCapture(
1176 self: *Daemon,
1177 gpa: std.mem.Allocator,
1178 client: *Client,
1179 term: *ghostty_vt.Terminal,
1180 payload: []const u8,
1181 ) !void {
1182 if (payload.len != @sizeOf(ipc.Capture)) return;
1183 const request = std.mem.bytesToValue(ipc.Capture, payload);
1184 const format = std.enums.fromInt(util.HistoryFormat, request.format) orelse return;
1185 self.setPwd(term);
1186 const output = util.serializeTerminalRange(gpa, term, format, request.rows);
1187 defer if (output) |text| gpa.free(text);
1188 try ipc.appendMessage(gpa, &client.write_buf, .Capture, output orelse "");
1189 client.has_pending_output = true;
1190 }
1191
1192 pub fn handleRun(self: *Daemon, gpa: std.mem.Allocator, io: std.Io, client: *Client, payload: []const u8) !void {
1193 // Reset task tracking so the new command's exit marker is detected.
1194 // Without this, a second `zmx run` on the same session is ignored
1195 // because task_exit_code is still set from the first run.
1196 self.task_exit_code = null;
1197 self.task_ended_at = null;
1198 self.is_task_mode = true;
1199 self.task_id = util.generateTaskId(io);
1200
1201 if (payload.len == 0) return;
1202
1203 const cmd = payload;
1204
1205 // Chain the exit marker with `;` on the same line. `$?` captures the
1206 // exit code of the command (not the `;`). The sole exception is when
1207 // the command contains a heredoc (`<<`), the delimiter must be alone
1208 // on its line, so the marker goes on the next line instead.
1209 var buf: [1024]u8 = undefined;
1210 const marker = try util.getTaskExitMarker(&buf, self.task_id);
1211 var single_buf: [1024]u8 = undefined;
1212 const single_line_marker = try std.fmt.bufPrint(&single_buf, "; echo {s}$?\r", .{marker});
1213 var here_buf: [1024]u8 = undefined;
1214 const heredoc_marker = try std.fmt.bufPrint(&here_buf, "\r\necho {s}$?\r", .{marker});
1215 const uses_heredoc = std.mem.indexOf(u8, cmd, "<<") != null;
1216
1217 if (cmd.len > 0 and cmd[cmd.len - 1] == '\r') {
1218 self.queuePtyInput(gpa, cmd[0 .. cmd.len - 1]);
1219 } else {
1220 self.queuePtyInput(gpa, cmd);
1221 }
1222 self.queuePtyInput(gpa, if (uses_heredoc) heredoc_marker else single_line_marker);
1223
1224 try ipc.appendMessage(gpa, &client.write_buf, .Ack, "");
1225 client.has_pending_output = true;
1226 std.log.debug("run command len={d}", .{payload.len});
1227 }
1228
1229 /// Store the session's working directory as a plain path.
1230 ///
1231 /// Accepts either an OSC 7 value (`file://<host><path>`, percent-encoded)
1232 /// or a path. Decoding here rather than at each use keeps `zmx list`
1233 /// printing a path and lets the chdir on session create find directories
1234 /// whose names needed escaping.
1235 ///
1236 /// The value is copied, so callers may pass a temporary.
1237 pub fn setCwd(self: *Daemon, value: []const u8) void {
1238 var buf: [std.fs.max_path_bytes]u8 = undefined;
1239 var host_buf: [std.posix.HOST_NAME_MAX]u8 = undefined;
1240 const hostname = std.posix.gethostname(&host_buf) catch "";
1241 const cwd = util.parseOsc7Cwd(&buf, value, hostname) orelse {
1242 std.log.warn("ignoring unusable cwd={s}", .{value});
1243 return;
1244 };
1245
1246 // Store the URI form. A caller that handed us a plain path gets one
1247 // built here, so `cwd` has the same shape no matter the source. A value
1248 // that already was a URI is kept verbatim, so `list` shows what the
1249 // shell actually reported.
1250 self.cwd = if (std.fs.path.isAbsolute(value))
1251 util.toOsc7Cwd(&self.cwd_buf, value, hostname) orelse return
1252 else blk: {
1253 if (value.len > self.cwd_buf.len) return;
1254 @memcpy(self.cwd_buf[0..value.len], value);
1255 break :blk self.cwd_buf[0..value.len];
1256 };
1257
1258 // Only keep an openable path when it names a directory on this host.
1259 if (cwd.is_local and cwd.path.len <= self.cwd_path_buf.len) {
1260 @memcpy(self.cwd_path_buf[0..cwd.path.len], cwd.path);
1261 self.cwd_path = self.cwd_path_buf[0..cwd.path.len];
1262 } else {
1263 self.cwd_path = "";
1264 }
1265 std.log.info("set cwd={s} path={s}", .{ self.cwd, self.cwd_path });
1266 }
1267
1268 fn setPwd(self: *Daemon, term: *ghostty_vt.Terminal) void {
1269 const pwd = term.getPwd() orelse return;
1270 if (std.mem.eql(u8, self.cwd, pwd)) return;
1271 self.setCwd(pwd);
1272 }
1273
1274 pub fn handleOutput(self: *Daemon, gpa: std.mem.Allocator, payload: []const u8, term: *ghostty_vt.Terminal, vt_stream: anytype) !void {
1275 vt_stream.nextSlice(payload);
1276 self.setPwd(term);
1277 self.has_pty_output = true;
1278 for (self.clients.items) |client| {
1279 try ipc.appendMessage(gpa, &client.write_buf, .Output, payload);
1280 client.has_pending_output = true;
1281 }
1282 if (self.clients.items.len > 0) {
1283 lib_posix.kill(self.pid, lib_posix.SIG.WINCH) catch |err| {
1284 std.log.warn("failed to send SIGWINCH err={s}", .{@errorName(err)});
1285 };
1286 }
1287 }
1288
1289 pub fn handleWrite(self: *Daemon, gpa: std.mem.Allocator, client: *Client, payload: []const u8) !void {
1290 // Wire format: [u32 path len][path bytes][file content]
1291 if (payload.len < @sizeOf(u32)) return error.InvalidPayload;
1292 const path_len = std.mem.bytesToValue(u32, payload[0..@sizeOf(u32)]);
1293 if (payload.len < @sizeOf(u32) + path_len) return error.InvalidPayload;
1294 const file_path = payload[@sizeOf(u32)..][0..path_len];
1295 const file_content = payload[@sizeOf(u32) + path_len ..];
1296
1297 // Inject file creation through the PTY so it works over SSH.
1298 // Base64-encode content and pipe through printf | base64 -d > file.
1299 // Chunk large files to stay under command-line length limits.
1300 // 48000 is divisible by 3 (clean base64 boundaries) and encodes
1301 // to ~64KB, well under typical ARG_MAX.
1302 // The commands skip queuePtyInput's cap: it would drop the chunks
1303 // past PTY_WRITE_BUF_MAX and leave the file cut short. The payload
1304 // is already bounded by ipc.MAX_PAYLOAD_LEN.
1305 const chunk_size = 48000;
1306 var offset: usize = 0;
1307 var is_first = true;
1308
1309 while (offset < file_content.len or is_first) {
1310 const end = @min(offset + chunk_size, file_content.len);
1311 const chunk = file_content[offset..end];
1312
1313 const encoded_len = std.base64.standard.Encoder.calcSize(chunk.len);
1314 const encoded = try gpa.alloc(u8, encoded_len);
1315 defer gpa.free(encoded);
1316 _ = std.base64.standard.Encoder.encode(encoded, chunk);
1317
1318 try self.pty_write_buf.appendSlice(gpa, "printf '%s' '");
1319 try self.pty_write_buf.appendSlice(gpa, encoded);
1320 if (is_first) {
1321 try self.pty_write_buf.appendSlice(gpa, "' | base64 -d > '");
1322 } else {
1323 try self.pty_write_buf.appendSlice(gpa, "' | base64 -d >> '");
1324 }
1325 try self.pty_write_buf.appendSlice(gpa, file_path);
1326 try self.pty_write_buf.appendSlice(gpa, "'");
1327 try self.pty_write_buf.appendSlice(gpa, "\r");
1328
1329 offset = end;
1330 is_first = false;
1331 }
1332
1333 try ipc.appendMessage(gpa, &client.write_buf, .Ack, "");
1334 client.has_pending_output = true;
1335 std.log.debug(
1336 "write command len={d} file_path={s}",
1337 .{ file_content.len, file_path },
1338 );
1339 }
1340
1341 fn handleEnvGet(self: *Daemon, gpa: std.mem.Allocator, client: *Client) !void {
1342 const leader_opt = self.getLeaderClient();
1343 const payload = if (leader_opt) |leader| leader.env_str orelse "" else "";
1344 try ipc.appendMessage(gpa, &client.write_buf, .EnvData, payload);
1345 client.has_pending_output = true;
1346 }
1347
1348 fn handleEnvSet(_: *Daemon, gpa: std.mem.Allocator, client: *Client, env_str: []const u8) !void {
1349 std.log.info("handle env set payload={s}", .{env_str});
1350 try client.setEnv(gpa, env_str);
1351
1352 try ipc.appendMessage(gpa, &client.write_buf, .Ack, "");
1353 client.has_pending_output = true;
1354 }
1355
1356 fn handleLabelGet(self: *Daemon, gpa: std.mem.Allocator, client: *Client) !void {
1357 const out = try label.labelsToU8(gpa, self.labels);
1358 defer gpa.free(out);
1359 try ipc.appendMessage(gpa, &client.write_buf, .LabelData, out);
1360 client.has_pending_output = true;
1361 }
1362
1363 fn handleLabelSet(self: *Daemon, gpa: std.mem.Allocator, client: *Client, labels: []const u8) !void {
1364 std.log.info("handle label set payload={s}", .{labels});
1365
1366 var kvs = label.LabelIterator.init(labels);
1367 while (kvs.next()) |kv| {
1368 if (kv.value.len == 0) {
1369 if (self.labels.fetchRemove(kv.key)) |existing| {
1370 gpa.free(existing.key);
1371 gpa.free(existing.value);
1372 }
1373 continue;
1374 }
1375
1376 const owned_key = try gpa.dupe(u8, kv.key);
1377 errdefer gpa.free(owned_key);
1378 const owned_value = try gpa.dupe(u8, kv.value);
1379 errdefer gpa.free(owned_value);
1380 if (try self.labels.fetchPut(gpa, owned_key, owned_value)) |existing| {
1381 // fetchPut does NOT replace the key in the map, the old
1382 // key pointer stays. So free the new (unused) key and the
1383 // old value.
1384 gpa.free(owned_key);
1385 gpa.free(existing.value);
1386 }
1387 }
1388
1389 try ipc.appendMessage(gpa, &client.write_buf, .Ack, "");
1390 client.has_pending_output = true;
1391 }
1392
1393 fn handleLabelClear(self: *Daemon, gpa: std.mem.Allocator, client: *Client) !void {
1394 var it = self.labels.iterator();
1395 while (it.next()) |entry| {
1396 gpa.free(entry.key_ptr.*);
1397 gpa.free(entry.value_ptr.*);
1398 }
1399 self.labels.clearRetainingCapacity();
1400 try ipc.appendMessage(gpa, &client.write_buf, .Ack, "");
1401 client.has_pending_output = true;
1402 }
1403};
1404
1405test "terminal retains the configured scrollback without the default byte cap" {
1406 const alloc = std.testing.allocator;
1407 const cfg = Cfg{ .socket_dir = "", .log_dir = "" };
1408 var term = try initTerminal(alloc, std.testing.io, .{ .cols = 80, .rows = 24 }, &cfg);
1409 defer term.deinit(alloc);
1410 var stream = term.vtStream();
1411 defer stream.deinit();
1412
1413 stream.nextSlice("first line\r\n");
1414 for (1..cfg.max_scrollback_lines) |_| stream.nextSlice("more output\r\n");
1415
1416 const history = util.serializeTerminal(alloc, &term, .plain) orelse return error.TestUnexpectedNull;
1417 defer alloc.free(history);
1418 try std.testing.expect(std.mem.startsWith(u8, history, "first line\n"));
1419
1420 // Exceed the limit comfortably because Ghostty prunes whole pages.
1421 for (0..cfg.max_scrollback_lines) |_| stream.nextSlice("more output\r\n");
1422 stream.nextSlice("latest line\r\n");
1423
1424 const pruned_history = util.serializeTerminal(alloc, &term, .plain) orelse return error.TestUnexpectedNull;
1425 defer alloc.free(pruned_history);
1426 try std.testing.expect(std.mem.indexOf(u8, pruned_history, "first line") == null);
1427 try std.testing.expect(std.mem.indexOf(u8, pruned_history, "latest line") != null);
1428}
1429
1430fn testDaemon() Daemon {
1431 return .{
1432 .cfg = undefined,
1433 .clients = .empty,
1434 .leader_client_fd = null,
1435 .session_name = "test",
1436 .socket_path = "",
1437 .running = true,
1438 .pid = 0,
1439 .created_at = 0,
1440 };
1441}
1442
1443fn testClient(alloc: std.mem.Allocator, fd: i32, is_terminal: bool) !*Client {
1444 const c = try alloc.create(Client);
1445 c.* = .{ .alloc = alloc, .socket_fd = fd, .read_buf = try ipc.SocketBuffer.init(alloc), .write_buf = .empty };
1446 c.is_terminal = is_terminal;
1447 return c;
1448}
1449
1450test "hasTerminalClient follows attach, detach and detach-all" {
1451 const alloc = std.testing.allocator;
1452 var daemon = testDaemon();
1453 defer daemon.clients.deinit(alloc);
1454 defer daemon.pty_write_buf.deinit(alloc);
1455
1456 // Real fds (pipes) so Client.deinit's close() is legal.
1457 const a = try lib_posix.pipe2(.{});
1458 const b = try lib_posix.pipe2(.{});
1459
1460 try std.testing.expect(!daemon.hasTerminalClient());
1461
1462 // A run/send/tail client doesn't count.
1463 const tail_client = try testClient(alloc, a[0], false);
1464 try daemon.clients.append(alloc, tail_client);
1465 try std.testing.expect(!daemon.hasTerminalClient());
1466
1467 // An attach client does, until it disconnects.
1468 const term_client = try testClient(alloc, a[1], true);
1469 try daemon.clients.append(alloc, term_client);
1470 try std.testing.expect(daemon.hasTerminalClient());
1471 _ = daemon.closeClient(alloc, tail_client, 0, false);
1472 try std.testing.expect(daemon.hasTerminalClient());
1473 _ = daemon.closeClient(alloc, term_client, 0, false);
1474 try std.testing.expect(!daemon.hasTerminalClient());
1475
1476 try daemon.clients.append(alloc, try testClient(alloc, b[0], true));
1477 try daemon.clients.append(alloc, try testClient(alloc, b[1], true));
1478 daemon.handleDetachAll(alloc);
1479 try std.testing.expect(!daemon.hasTerminalClient());
1480 try std.testing.expectEqual(@as(?i32, null), daemon.leader_client_fd);
1481}
1482
1483test "send queues PTY input without changing leader" {
1484 const alloc = std.testing.allocator;
1485 var daemon = testDaemon();
1486 daemon.leader_client_fd = 42;
1487 defer daemon.pty_write_buf.deinit(alloc);
1488
1489 daemon.handleSend(alloc, "hello");
1490
1491 try std.testing.expectEqual(@as(?i32, 42), daemon.leader_client_fd);
1492 try std.testing.expectEqualStrings("hello", daemon.pty_write_buf.items);
1493}
1494
1495test "handleEnvGet returns leader client's environment variables including unsets" {
1496 const alloc = std.testing.allocator;
1497 var cfg = Cfg{
1498 .socket_dir = "/tmp",
1499 .log_dir = "/tmp",
1500 };
1501
1502 var daemon = Daemon{
1503 .cfg = &cfg,
1504 .clients = .empty,
1505 .session_name = "test",
1506 .socket_path = try alloc.dupe(u8, ""),
1507 .running = true,
1508 .pid = 0,
1509 .created_at = 0,
1510 };
1511 defer daemon.deinit(alloc);
1512
1513 const fds1 = try lib_posix.pipe2(.{});
1514 defer lib_posix.close(fds1[1]);
1515 var client1 = Client{
1516 .alloc = alloc,
1517 .socket_fd = fds1[0],
1518 .read_buf = try ipc.SocketBuffer.init(alloc),
1519 .write_buf = std.ArrayList(u8).empty,
1520 };
1521 defer client1.deinit(alloc);
1522 try client1.setEnv(alloc, "DISPLAY=:1\nSSH_AUTH_SOCK=/tmp/ssh-1\nKITTY_LISTEN_ON=unix:/tmp/kitty-1\n-WINDOWID\n");
1523
1524 const fds2 = try lib_posix.pipe2(.{});
1525 defer lib_posix.close(fds2[1]);
1526 var client2 = Client{
1527 .alloc = alloc,
1528 .socket_fd = fds2[0],
1529 .read_buf = try ipc.SocketBuffer.init(alloc),
1530 .write_buf = std.ArrayList(u8).empty,
1531 };
1532 defer client2.deinit(alloc);
1533 try client2.setEnv(alloc, "SSH_AUTH_SOCK=/tmp/ssh-2\nWINDOWID=12345\nKITTY_LISTEN_ON=unix:/tmp/kitty-2\n-DISPLAY\n");
1534
1535 const fds_req = try lib_posix.pipe2(.{});
1536 defer lib_posix.close(fds_req[1]);
1537 var client_req = Client{
1538 .alloc = alloc,
1539 .socket_fd = fds_req[0],
1540 .read_buf = try ipc.SocketBuffer.init(alloc),
1541 .write_buf = std.ArrayList(u8).empty,
1542 };
1543 defer client_req.deinit(alloc);
1544
1545 try daemon.clients.append(alloc, &client1);
1546 try daemon.clients.append(alloc, &client2);
1547
1548 // No leader yet -> handleEnvGet returns empty string payload
1549 try daemon.handleEnvGet(alloc, &client_req);
1550 try std.testing.expect(client_req.write_buf.items.len > 0);
1551 client_req.write_buf.clearRetainingCapacity();
1552
1553 // Set client1 as leader
1554 try daemon.setLeader(alloc, &client1);
1555 try std.testing.expectEqual(@as(?i32, fds1[0]), daemon.leader_client_fd);
1556 try daemon.handleEnvGet(alloc, &client_req);
1557 // Wire message: [Header][Payload]
1558 const pay1 = client_req.write_buf.items[@sizeOf(ipc.Header)..];
1559 try std.testing.expectEqualStrings("DISPLAY=:1\nSSH_AUTH_SOCK=/tmp/ssh-1\nKITTY_LISTEN_ON=unix:/tmp/kitty-1\n-WINDOWID\n", pay1);
1560 client_req.write_buf.clearRetainingCapacity();
1561
1562 // Switch leader to client2
1563 try daemon.setLeader(alloc, &client2);
1564 try std.testing.expectEqual(@as(?i32, fds2[0]), daemon.leader_client_fd);
1565 try daemon.handleEnvGet(alloc, &client_req);
1566 const pay2 = client_req.write_buf.items[@sizeOf(ipc.Header)..];
1567 try std.testing.expectEqualStrings("SSH_AUTH_SOCK=/tmp/ssh-2\nWINDOWID=12345\nKITTY_LISTEN_ON=unix:/tmp/kitty-2\n-DISPLAY\n", pay2);
1568 client_req.write_buf.clearRetainingCapacity();
1569
1570 // Client2 updates env
1571 try daemon.handleEnvSet(alloc, &client2, "DISPLAY=:99\n");
1572 try daemon.handleEnvGet(alloc, &client_req);
1573 const pay3 = client_req.write_buf.items[@sizeOf(ipc.Header)..];
1574 try std.testing.expectEqualStrings("DISPLAY=:99\n", pay3);
1575}