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