//! The terminal shell: owns the event loop and all IO. Translates vaxis
//! events into core events, performs the core's effects (fork ptys, write
//! them, resize them), and hands the core's Surface to vaxis cell-for-cell —
//! the canonical interface rendered with no interpretation.
const std = @import("std");
const builtin = @import("builtin");
const posix = std.posix;
const libc = std.c;
const linux = std.os.linux;
const vaxis = @import("vaxis");
const pardes = @import("../pardes.zig");
const tracy = @import("../tracy.zig");
const look = @import("../look.zig");
const shell_bin = @import("../shell_bin.zig");
const message = @import("../message.zig");
const file_watch = @import("../file_watch.zig");
const user_config = @import("../user_config.zig");
const selection_pipe = @import("../selection_pipe.zig");
const nested = @import("../nested.zig");
const fuse = @import("../fuse.zig");
const fs_service = @import("../fs_service.zig");
const panel_compositor = @import("panel_compositor.zig");
const host_api = @import("../host.zig");
const config = @import("../config.zig");
// The other half of `--detach`, and the reason this file has an `--attach`
// branch at all: the frontend side of a detached session is a terminal and a
// socket, and this file is already the one that owns a terminal.
const detached_client = @import("../detached/client.zig");
const detached_server = @import("../detached/server.zig");
const wire = @import("../detached/wire.zig");

extern "c" fn forkpty(amaster: *c_int, name: ?[*:0]u8, termp: ?*const anyopaque, winp: ?*const posix.winsize) c_int;
extern "c" fn execv(path: [*:0]const u8, argv: [*:null]const ?[*:0]const u8) c_int;
extern "c" fn chdir(path: [*:0]const u8) c_int;
extern "c" fn _exit(status: c_int) noreturn;
extern "c" fn setenv(name: [*:0]const u8, value: [*:0]const u8, overwrite: c_int) c_int;

// TIOCSWINSZ: absent from std.c.T on darwin — _IOW('t', 103, winsize)
const TIOCSWINSZ: c_int = @bitCast(@as(u32, if (@hasDecl(posix.T, "IOCSWINSZ")) posix.T.IOCSWINSZ else 0x80087467));

pub const Command = struct {
    pub var value: union(enum) {
        nop,
        quit,
        tick,
        key_press: vaxis.Key,
        pty_read: struct { id: usize, bytes: []u8 },
        pty_eof: struct { id: usize, gen: u32 },
        winsize: vaxis.Winsize,
        mouse: vaxis.Mouse,
        /// Focus reporting is part of vaxis's mouse mode (DEC 1004). A TTY
        /// cannot report a literal pointer crossing its character grid, so
        /// losing terminal focus is its only reliable pointer-leave signal.
        focus_in,
        focus_out,
        paste: []const u8,
        /// The bracketed-paste brackets. vaxis posts them ONLY because this
        /// union declares fields with these exact names — its Loop gates every
        /// event on `@hasField` — and the pasted bytes themselves arrive
        /// BETWEEN them as ordinary key presses, which the loop accumulates
        /// into one `.paste` above instead of running as commands.
        paste_start,
        paste_end,
        /// a language query finished on a worker; rows are lsp-domain-owned
        lsp_done: struct { id: u32, rows: []u8 },
        /// a selection-filter worker finished; every stdout is gpa-owned
        pipe_done: selection_pipe.Response,
        /// something happened in a watched directory (see watchFiles)
        files_changed,
        /// a pardes launched inside this one sent us a builtin command line
        /// (see lookServer); gpa-owned, like pty_read
        command: []u8,
        /// `--fs`: the /dev/fuse descriptor has requests on it. Carries
        /// nothing and is applied as a no-op — its whole job is to end the
        /// blocking `nextEvent`, because the drain itself lives in pollFrame
        /// beside the file-watch reload. Same shape and same reason as
        /// `files_changed` above, and posted from two places: the poll thread
        /// when the kernel makes the descriptor readable, and pollFrame itself
        /// when a batch hit its cap with requests still pending.
        fs_ready,
    } = .nop;
};
const Loop = vaxis.Loop(@TypeOf(Command.value));

/// One language query, owned by the worker that runs it. Everything the
/// backend may read is copied in here before the worker starts: the core goes
/// on editing the moment the effect is drained, so a borrowed slice would be a
/// use-after-free the length of one keystroke.
const LspJob = struct {
    id: u32,
    kind: pardes.lsp.Kind,
    offset: u32,
    path: []u8,
    source: [:0]u8,
    arg: []u8,
    root: []u8,

    fn free(j: *LspJob, allocator: std.mem.Allocator) void {
        allocator.free(j.path);
        allocator.free(j.source);
        allocator.free(j.arg);
        allocator.free(j.root);
        allocator.destroy(j);
    }
};

const max_pipe_tasks = 16;

const PipeTask = struct {
    id: u32,
    future: std.Io.Future(anyerror!void),
};

const PipeTasks = struct {
    items: [max_pipe_tasks]PipeTask = undefined,
    len: usize = 0,

    fn full(tasks: *const PipeTasks) bool {
        return tasks.len == tasks.items.len;
    }

    fn add(tasks: *PipeTasks, task: PipeTask) void {
        std.debug.assert(!tasks.full());
        tasks.items[tasks.len] = task;
        tasks.len += 1;
    }

    fn finish(tasks: *PipeTasks, io: std.Io, id: u32) void {
        for (tasks.items[0..tasks.len], 0..) |*task, i| if (task.id == id) {
            task.future.await(io) catch {};
            tasks.len -= 1;
            std.mem.copyForwards(PipeTask, tasks.items[i..tasks.len], tasks.items[i + 1 .. tasks.len + 1]);
            return;
        };
    }

    fn cancelAll(tasks: *PipeTasks, io: std.Io) void {
        for (tasks.items[0..tasks.len]) |*task| task.future.cancel(io) catch {};
        tasks.len = 0;
    }
};

const Pty = struct {
    file: std.Io.File,
    pid: posix.pid_t,
    reader: std.Io.Future(anyerror!void),
};

const KittyPlacement = struct {
    cell_x: u16,
    cell_y: u16,
    cell_cols: u16,
    cell_rows: u16,
    options: vaxis.Image.DrawOptions,
};

fn terminalCellPixels(total_pixels: u16, cells: u16, fallback: u32) u32 {
    if (total_pixels == 0 or cells == 0) return fallback;
    return @max(1, @as(u32, total_pixels) / cells);
}

fn updateCoreTerminalSize(core: *pardes.Pardes, cols: u16, rows: u16, pixel_w: u16, pixel_h: u16) void {
    core.update(.{ .resize = .{
        .cols = cols,
        .rows = rows,
        .cell_pixels = .{
            .w = @intCast(terminalCellPixels(pixel_w, cols, 8)),
            .h = @intCast(terminalCellPixels(pixel_h, rows, 16)),
        },
    } });
}

/// Translate backend-neutral source/destination pixels into Kitty's source
/// crop plus cell-sized placement. Kitty can specify only one scaled axis
/// without distorting the image; the terminal derives the other axis.
fn kittyPlacement(
    place: pardes.ImagePlace,
    screen_cols: u16,
    screen_rows: u16,
    screen_pixel_w: u16,
    screen_pixel_h: u16,
) ?KittyPlacement {
    if (comptime pardes.pdf_enabled) {
        if (place.native.fit == .contain) return null;
        const cell_w = terminalCellPixels(screen_pixel_w, screen_cols, 8);
        const cell_h = terminalCellPixels(screen_pixel_h, screen_rows, 16);
        const body_w = std.math.mul(u32, place.w, cell_w) catch return null;
        const body_h = std.math.mul(u32, place.h, cell_h) catch return null;
        const geometry = place.native.geometry orelse pardes.image.nativeGeometry(
            place.iw,
            place.ih,
            body_w,
            body_h,
            place.native.fit,
            place.native.pan_x,
            place.native.pan_y,
        ) orelse return null;

        // Kitty has a top-left pixel offset but no destination bottom clip.
        // Its missing c/r axis is rounded up to whole terminal cells, so a
        // clipped fragment shorter than one row cannot be represented without
        // painting the following theme gap. Conservatively keep only whole
        // rows contained by geometry.dst and trim the source crop to the same
        // scale. Cached page pixels remain unchanged; an unrepresentable tail
        // is simply left as theme background.
        const safe_rows_u32 = geometry.dst.h / cell_h;
        if (safe_rows_u32 == 0) return null;
        const safe_pixel_h = safe_rows_u32 * cell_h;
        var safe_src_h: u32 = geometry.src.h;
        var declared_cols: u32 = 0;
        var declared_rows: u32 = 0;
        switch (place.native.fit) {
            .width => {
                declared_cols = @intCast(@max(
                    @as(u64, 1),
                    (@as(u64, geometry.dst.w) + cell_w - 1) / cell_w,
                ));
                const declared_pixel_w = @as(u64, declared_cols) * cell_w;
                const max_src_h: u32 = @intCast(
                    @as(u64, safe_pixel_h) * geometry.src.w / declared_pixel_w,
                );
                if (max_src_h == 0) return null;
                safe_src_h = @min(safe_src_h, max_src_h);
                const aspect_pixel_h: u32 = @intCast(
                    (@as(u64, safe_src_h) * declared_pixel_w + geometry.src.w - 1) /
                        geometry.src.w,
                );
                declared_rows = @intCast(
                    (@as(u64, aspect_pixel_h) + cell_h - 1) / cell_h,
                );
            },
            .height => {
                declared_rows = safe_rows_u32;
                safe_src_h = @max(@as(u32, 1), @as(u32, @intCast(
                    @as(u64, safe_pixel_h) * geometry.src.h / geometry.dst.h,
                )));
                safe_src_h = @min(safe_src_h, geometry.src.h);
                const aspect_pixel_w: u32 = @intCast(
                    (@as(u64, geometry.src.w) * safe_pixel_h + safe_src_h - 1) /
                        safe_src_h,
                );
                declared_cols = @intCast(
                    (@as(u64, aspect_pixel_w) + cell_w - 1) / cell_w,
                );
            },
            .contain => unreachable,
        }
        if (declared_rows == 0 or declared_rows > safe_rows_u32) return null;

        // Every Kitty protocol field is u16. Reject an attachment which the
        // wire format cannot represent instead of truncating it.
        const src_x = std.math.cast(u16, geometry.src.x) orelse return null;
        const src_y = std.math.cast(u16, geometry.src.y) orelse return null;
        const src_w = std.math.cast(u16, geometry.src.w) orelse return null;
        const src_h = std.math.cast(u16, safe_src_h) orelse return null;
        const cell_x = std.math.cast(u16, geometry.dst.x / cell_w) orelse return null;
        const cell_y = std.math.cast(u16, geometry.dst.y / cell_h) orelse return null;
        if (cell_x >= place.w or cell_y >= place.h) return null;
        const pixel_x = std.math.cast(u16, geometry.dst.x % cell_w) orelse return null;
        const pixel_y = std.math.cast(u16, geometry.dst.y % cell_h) orelse return null;
        const cell_cols = std.math.cast(u16, @min(
            @as(u32, place.w - cell_x),
            (@as(u64, pixel_x) + @as(u64, declared_cols) * cell_w + cell_w - 1) / cell_w,
        )) orelse return null;
        const cell_rows = std.math.cast(u16, @min(
            @as(u32, place.h - cell_y),
            (@as(u64, pixel_y) + declared_rows * cell_h + cell_h - 1) / cell_h,
        )) orelse return null;
        if (cell_cols == 0 or cell_rows == 0) return null;

        return .{
            .cell_x = cell_x,
            .cell_y = cell_y,
            .cell_cols = cell_cols,
            .cell_rows = cell_rows,
            .options = .{
                .clip_region = .{ .x = src_x, .y = src_y, .width = src_w, .height = src_h },
                .pixel_offset = if (pixel_x != 0 or pixel_y != 0) .{ .x = pixel_x, .y = pixel_y } else null,
                .size = switch (place.native.fit) {
                    .width => .{ .cols = std.math.cast(u16, declared_cols) orelse return null },
                    .height => .{ .rows = std.math.cast(u16, declared_rows) orelse return null },
                    .contain => unreachable,
                },
            },
        };
    } else {
        return null;
    }
}

test "Kitty PDF fragments never declare pixels beyond their clipped bottom" {
    if (comptime !pardes.pdf_enabled) return;

    const base = pardes.ImagePlace{
        .pane = 0,
        .serial = 1,
        .native = .{
            .revision = 1,
            .page = 0,
            .fit = .width,
            .geometry = .{
                .src = .{ .w = 96, .h = 10 },
                .dst = .{ .y = 195, .w = 304, .h = 29 },
            },
        },
        .x = 0,
        .y = 0,
        .w = 38,
        .h = 14,
        .rgba = &.{},
        .iw = 96,
        .ih = 64,
    };
    const width = kittyPlacement(base, 80, 16, 640, 256) orelse
        return error.MissingSafeKittyWidthFragment;
    const width_clip = width.options.clip_region.?;
    const width_size = width.options.size.?;
    try std.testing.expectEqual(@as(u16, 5), width_clip.height.?);
    try std.testing.expectEqual(@as(u16, 38), width_size.cols.?);
    try std.testing.expectEqual(@as(u16, 3), width.options.pixel_offset.?.y);
    const aspect_pixels = (@as(u32, width_size.cols.?) * 8 * width_clip.height.? +
        width_clip.width.? - 1) / width_clip.width.?;
    const inferred_rows = (aspect_pixels + 15) / 16;
    try std.testing.expect(inferred_rows * 16 <= base.native.geometry.?.dst.h);

    var height = base;
    height.native.fit = .height;
    height.native.geometry.?.src.h = 9;
    const height_fragment = kittyPlacement(height, 80, 16, 640, 256) orelse
        return error.MissingSafeKittyHeightFragment;
    try std.testing.expectEqual(@as(u16, 1), height_fragment.options.size.?.rows.?);
    try std.testing.expect(@as(u32, height_fragment.options.size.?.rows.?) * 16 <=
        height.native.geometry.?.dst.h);

    // There is no honest APC for less than one physical row: omitting it is
    // preferable to painting three pixels of the following theme gap.
    height.native.geometry.?.dst.h = 13;
    try std.testing.expect(kittyPlacement(height, 80, 16, 640, 256) == null);
}

test "Linux host watch closes initial race and reloads rename-over PDF while idle" {
    if (comptime builtin.os.tag != .linux or !pardes.pdf_enabled) return;
    const io = std.testing.io;
    const gpa = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const original = try pardes.pdf.makeOutlineTestPdf(gpa);
    defer gpa.free(original);
    const replacement = try pardes.pdf.makeNoOutlineTestPdf(gpa);
    defer gpa.free(replacement);
    try tmp.dir.writeFile(io, .{ .sub_path = "live.pdf", .data = original });
    // Prepare both editor-style temporary inodes before marking the directory
    // so the only post-arm wake below is the second rename.
    try tmp.dir.writeFile(io, .{ .sub_path = "initial.pdf", .data = replacement });
    try tmp.dir.writeFile(io, .{ .sub_path = "live-replacement.pdf", .data = original });
    var path_buf: [256]u8 = undefined;
    const path = try std.fmt.bufPrint(&path_buf, ".zig-cache/tmp/{s}/live.pdf", .{tmp.sub_path});

    const fd = libc.inotify_init1(linux.IN.CLOEXEC | linux.IN.NONBLOCK);
    if (fd < 0) return error.InotifyInitFailed;
    defer _ = libc.close(fd);
    var watches: file_watch.Table = @splat(null);
    defer for (0..pardes.MAX_PANES) |id| file_watch.watchPane(
        fd,
        &watches,
        @intCast(id),
        null,
        0,
        .{ .text = 0 },
    );

    const core = try pardes.Pardes.init(gpa, .{ .file = path, .cols = 80, .rows = 28 });
    defer core.deinit();
    try std.testing.expectEqual(@as(usize, 3), core.panes[0].?.pdf.?.page_count);

    // The core opened the three-page inode, but the host has not drained its
    // watch effect yet. Replace it now: install-then-reconcile must discover
    // the one-page document even though no source existed for this first edge.
    try tmp.dir.rename("initial.pdf", tmp.dir, "live.pdf", io);
    var armed = false;
    while (core.nextEffect()) |effect| switch (effect) {
        .watch => |watch| if (watch.pane == 0 and watch.on) {
            armed = true;
            try std.testing.expect(!file_watch.applyEffect(core, io, gpa, fd, &watches, 0, true));
        },
        else => {},
    };
    try std.testing.expect(armed and watches[0] != null);
    try std.testing.expectEqual(@as(usize, 1), core.panes[0].?.pdf.?.page_count);

    try tmp.dir.rename("live-replacement.pdf", tmp.dir, "live.pdf", io);
    var events: [4096]u8 = undefined;
    const event_bytes = libc.read(fd, &events, events.len);
    try std.testing.expect(event_bytes > 0);

    // This is the same pass the watcher thread schedules; no key, mouse, or
    // synthetic core file_changed event participates in the transaction.
    try std.testing.expect(!file_watch.reloadChanged(core, io, gpa, &watches));
    const pane = core.panes[0].?;
    try std.testing.expectEqual(@as(usize, 3), pane.pdf.?.page_count);
    const disk_identity = try file_watch.identify(io, path);
    switch (watches[0].?.generation) {
        .pdf => |accepted| try std.testing.expect(accepted != null and accepted.?.eql(disk_identity)),
        .text => return error.PdfWatchStoredTextGeneration,
    }
    try std.testing.expect(std.mem.indexOf(u8, pane.msg[0..pane.msg_len], "reloaded") != null);

    core.native_images = true;
    var frame = std.heap.ArenaAllocator.init(gpa);
    defer frame.deinit();
    const surface = try core.render(frame.allocator());
    try std.testing.expect(surface.nimages > 0);
    try std.testing.expectEqual(@as(u32, 0), surface.images[0].?.native.page);
}

fn kittyImageRepresentable(place: pardes.ImagePlace) bool {
    return place.iw > 0 and place.ih > 0 and
        place.iw <= std.math.maxInt(u16) and place.ih <= std.math.maxInt(u16);
}

fn surfaceHasKittyKey(surface: *const pardes.Surface, key: pardes.ImageCacheKey) bool {
    for (surface.images[0..surface.nimages]) |maybe| {
        const place = maybe orelse continue;
        if (comptime pardes.pdf_enabled) if (!kittyImageRepresentable(place)) continue;
        if (place.cacheKey().eql(key)) return true;
    }
    return false;
}

const PdfWheelTarget = struct { pane: usize, page: usize };

/// A native PDF's page geometry is invalid between `setPdfPage` and the next
/// render. Remember the page under a vertical wheel press so the input batch
/// can stop exactly when that press crosses a page boundary. The following
/// queued wheel report then sees the newly rastered page instead of treating
/// missing geometry as another page-wise fallback.
fn nativePdfWheelTarget(core: *const pardes.Pardes, mouse: vaxis.Mouse) ?PdfWheelTarget {
    if (comptime !pardes.pdf_enabled) return null;
    if (!core.native_images or mouse.type != .press or
        (mouse.button != .wheel_up and mouse.button != .wheel_down) or
        mouse.col < 0 or mouse.row < 0) return null;
    const col: u16 = @intCast(mouse.col);
    const row: u16 = @intCast(mouse.row);
    for (core.panes, 0..) |slot, id| {
        const pane = slot orelse continue;
        const rect = core.rects[id];
        if (col < rect.x or col >= rect.x + rect.w or
            row < rect.y or row >= rect.y + rect.h) continue;
        const page = pane.pdfPage() orelse return null;
        return .{ .pane = id, .page = page };
    }
    return null;
}

/// `attach` is `--attach[=<name>]`: empty means "the session there is" (see
/// `sessionName`). It is a parameter rather than an `Options` field because it
/// says nothing to the core — this process does not have one when it is set.
pub fn run(init: std.process.Init, opts: pardes.Options, attach: ?[]const u8) !void {
    const io = init.io;
    const gpa = init.gpa;

    // `--attach` only, and registered HERE — before the terminal is opened —
    // for the LIFO: it must run after every deferred restore below. Why a
    // frontend stopped is discovered deep inside the loop while the alt screen
    // is still up, and anything written there is erased by the switch back to
    // the main screen, which is the one screen a user would look at.
    var attach_end: AttachEnd = .none;
    defer attach_end.report(io);

    // ...and resolved before the terminal too, for the same reason turned the
    // other way: "no session called work" is a launch that never started, and
    // flashing the alt screen up and straight back down to say so is worse
    // than never entering it.
    var name_buf: [detached_server.path_max]u8 = undefined;
    var attach_name: []const u8 = &.{};
    if (attach) |requested| attach_name = sessionName(&name_buf, requested, &attach_end) orelse return;

    const allocs = pardes.allocators.init(gpa);
    defer pardes.allocators.deinit();
    var options = opts;
    options.image_allocator = allocs.image;
    options.pdf_allocator = allocs.pdf;
    options.tree_sitter_allocator = allocs.tree_sitter;
    // the 16 MiB static buffer behind every per-frame Surface
    options.frame_allocator = allocs.frame;

    // SIGWINCH must never run vaxis's signal handler: it posts the winsize
    // event through std.Io.Mutex/Condition, and when the signal lands on a
    // thread blocked inside an Io.Threaded syscall region (pty readers in
    // read(2), the main thread parked in queue.pop) a contended lock re-enters
    // the Io machinery and Syscall.start hits `unreachable` — panic, then the
    // panic-time terminal restore used to write through the same Io and
    // recurse until stack overflow. Reproduced by resizing the outer terminal
    // (e.g. a font-size change) while shells run. Block it here, before any
    // thread exists (threads inherit the mask, so vaxis's handler never
    // fires), and take it synchronously on the sigwait thread below instead.
    var winch_set = posix.sigemptyset();
    posix.sigaddset(&winch_set, posix.SIG.WINCH);
    posix.sigprocmask(posix.SIG.BLOCK, &winch_set, null);

    var tty_buf: [0x10000]u8 = undefined;
    var tty = try vaxis.Tty.init(io, &tty_buf);
    defer tty.deinit(); // restore cooked termios LAST, after vx flushed its resets
    var vx = try vaxis.init(io, gpa, init.environ_map, .{ .system_clipboard_allocator = gpa });
    defer vx.deinit(gpa, tty.writer());
    try vx.enterAltScreen(tty.writer());
    defer vx.exitAltScreen(tty.writer()) catch {};
    // requests 1002;1003;1004;1006 (cell-coordinate SGR; called pre-query, so
    // vaxis never upgrades to 1016 pixel mode). Note: ghostty's GTK apprt drops
    // middle press+release BEFORE mouse reporting when the desktop sets
    // gtk-enable-primary-paste=false — no mode we request can surface middle
    // clicks there (see test/snapshots/ghostty-mid.snap).
    try vx.setMouseMode(tty.writer(), true);
    // Bracketed paste. Without it a paste into pardes-in-a-terminal is just a
    // flood of key presses: plausible-looking in insert mode, and in normal
    // mode every pasted character runs as a command. With it the terminal
    // wraps the bytes in \x1b[200~ / \x1b[201~ and the loop coalesces them.
    // No defer to switch it back off, for the same reason the mouse modes
    // above have none: setBracketedPaste records state.bracketed_paste, and
    // vaxis's resetState — reached from the `defer vx.deinit` above, while the
    // tty is still open — sends the disable off that flag.
    try vx.setBracketedPaste(tty.writer(), true);

    // `--attach`: this process has a terminal and NO core. Everything above is
    // the terminal, which an attached frontend needs exactly as much as a whole
    // session does; everything below is the core, which lives in the detached
    // process. The branch is here so both leave by the same door — an attach
    // has to restore cooked mode, the main screen and the mouse the way an
    // ordinary exit does, and sharing the deferred teardown is the only way to
    // guarantee that instead of asserting it.
    if (attach != null) {
        attach_end = attachSession(init, opts, attach_name, &tty, &vx);
        return;
    }

    pardes.image.start(io, allocs.image);
    if (comptime pardes.pdf_enabled) pardes.pdf.start(allocs.pdf);
    pardes.syntax.start(allocs.tree_sitter);
    defer {
        pardes.image.stop();
        if (comptime pardes.pdf_enabled) pardes.pdf.stop();
        pardes.syntax.stop();
    }

    var core = if (options.load_path) |lp| blk: {
        const bytes = try look.readFile(gpa, lp);
        defer gpa.free(bytes);
        break :blk try pardes.Pardes.initFromDump(allocs.pardes, options, bytes);
    } else try pardes.Pardes.init(allocs.pardes, options);
    defer core.deinit();

    // Private, complete before any fork and retained until the last possible
    // spawn; children borrow only these stable in-struct path buffers.
    var prompt_rcs = shell_bin.PromptRcs.init();
    defer prompt_rcs.deinit();
    // macos: apple's bash 3.2 prints the zsh-deprecation banner into every
    // pane unless this is in the env BEFORE bash starts (the rc is too late)
    if (comptime builtin.os.tag.isDarwin()) _ = setenv("BASH_SILENCE_DEPRECATION_WARNING", "1", 1);

    var kitty_handles = std.AutoHashMap(pardes.ImageCacheKey, vaxis.Image).init(gpa);
    defer {
        var iterator = kitty_handles.valueIterator();
        while (iterator.next()) |handle| vx.freeImage(tty.writer(), handle.id);
        kitty_handles.deinit();
    }
    var frame_arena: std.heap.ArenaAllocator = .init(allocs.frame);
    defer frame_arena.deinit();
    var paste_buf: std.Io.Writer.Allocating = .init(gpa);
    defer paste_buf.deinit();
    var loop: Loop = .init(io, &tty, &vx);

    // `--fs`: mount before the initial spawns, because those shells are the
    // ones that need PARDES_FS in their environment, and before the first
    // frame, because a script racing startup must find panes that are already
    // there. Also before any thread of ours exists — the mount forks the
    // setuid fusermount3 helper, and forking from a multithreaded process is
    // the hazard this whole region is ordered around. Null covers both "no
    // --fs" and "--fs but the mount failed"; the second is reported on a
    // message row inside `start` and the session runs on regardless.
    //
    // The teardown answers every held request, aborts the connection,
    // unmounts and removes `<parent>/<pid>`. The PARENT (`.../pardes`) stays,
    // like nested.zig's socket directory: another session may be living in it,
    // and rmdir of a shared directory is not ours to attempt.
    var fs = fs_service.start(gpa, core);
    // Covers the error paths only: the ordinary exit unmounts at the END OF
    // THE LOOP instead, see there.
    defer if (fs) |f| f.deinit();

    var sh: Shell = .{
        .io = io,
        .gpa = gpa,
        .lsp_gpa = allocs.lsp,
        .core = core,
        .prompt_rcs = &prompt_rcs,
        .loop = &loop,
        .vx = &vx,
        .tty = &tty,
        .kitty = &kitty_handles,
        .frame = &frame_arena,
        .paste_buf = &paste_buf,
        // One inotify instance for every watched pane, opened here — before
        // any thread exists — so the pre-loop effect drain below can already
        // mark the file a positional path argument opened. -1 off linux:
        // watchPane goes quiet and the core simply never gets a file_changed.
        .inotify_fd = if (builtin.os.tag == .linux) libc.inotify_init1(linux.IN.CLOEXEC) else -1,
        .fs = fs,
    };
    defer {
        // reap the reader tasks (cancel interrupts a blocked read) before
        // closing the masters — the runtime joins those threads on exit and a
        // reader stuck in read(2) would hang the process — then drain the
        // queue: leftover events own gpa bytes and would dump as leaks.
        for (&sh.ptys) |*slot| if (slot.*) |*pt| {
            pt.reader.cancel(io) catch {};
            _ = libc.close(pt.file.handle);
            slot.* = null;
        };
        // join the query worker BEFORE the drain below, or its late post
        // lands in a queue nobody empties again and the rows leak
        if (sh.lsp_task) |*t| {
            t.cancel(io) catch {};
            sh.lsp_task = null;
        }
        sh.pipe_tasks.cancelAll(io);
        // same contract as the pty readers: cancel unblocks the watcher's
        // read, and only then is the fd safe to close
        if (sh.watch_task) |*t| {
            t.cancel(io) catch {};
            sh.watch_task = null;
        }
        if (sh.inotify_fd >= 0) {
            _ = libc.close(sh.inotify_fd);
            sh.inotify_fd = -1;
        }
        while (loop.tryEvent() catch null) |ev| switch (ev) {
            .pty_read => |pr| gpa.free(pr.bytes),
            .command => |line| gpa.free(line),
            .paste => |b| gpa.free(@constCast(b)),
            .lsp_done => |d| allocs.lsp.free(d.rows),
            .pipe_done => |response_value| {
                var response = response_value;
                response.deinit(gpa);
            },
            else => {},
        };
    }
    const host = sh.host();

    // The socket a pardes launched inside this one connects to (nested.zig).
    // Declared AFTER the drain above so its teardown runs BEFORE it — the
    // listener thread must be out of the way before the queue is emptied.
    // --nested opted out of the whole mechanism, including being an outer
    // instance; so does any failure to bind, and then children simply open
    // their own session.
    const sock_fd: c_int = if (options.nested) -1 else nested.listen();
    defer nested.unlisten(sock_fd);

    // Perform the initial spawns BEFORE any worker thread exists: forkpty from
    // a multithreaded process can wedge the child before exec. `pump` installs
    // the host on every pass; this drain runs outside it, so install it here.
    core.host = host;
    while (core.nextEffect()) |effect| core.perform(effect);

    try loop.start();
    defer loop.stop();
    // resize watcher: plain detached thread (not io.concurrent — teardown
    // joins those, and sigwait never returns); dies with the process
    (try std.Thread.spawn(.{}, winchWatch, .{ &loop, &vx, &tty })).detach();
    // ...and the nested-instance listener, detached for the same reason: a
    // blocking accept(2) never returns either, so an io.concurrent task would
    // hang the teardown that joins it.
    if (sock_fd >= 0) (try std.Thread.spawn(.{}, lookServer, .{ gpa, sock_fd, &loop })).detach();
    // ...and the /dev/fuse poller, which is the same kind of thread again: it
    // waits for POLLIN and posts, never touching the core or the descriptor's
    // data. Joined by `Fs.deinit` rather than detached, because unlike accept4
    // it CAN be woken — fuse.zig gives it a control pipe for exactly that.
    fs_service.wake(fs, &loop, wakeFs);
    // Capability handshake — SEND the probes, do not wait on them. This was
    // queryTerminal(2ms), which blocks on a futex until DA1 comes back. The
    // number has to beat one terminal round trip: a local terminal answers in
    // microseconds, `ssh localhost` in under 1ms, and any real link never.
    // Measured over sshd on :22 with the replies delayed to model the wire,
    // 2ms already loses at 5ms RTT and everything above.
    //
    // Losing it is worse than never probing, because vaxis splits detect from
    // enable and only detect respects the deadline. queryTerminal sets
    // queries_done the moment the futex times out, and the two replies vaxis
    // gates on that flag — explicit width and scaled text, both answered as a
    // cursor-position report — stop being recognised as probe replies and are
    // handed to US as shift-F3/alt-F3 keypresses. The replies it does NOT
    // gate (mode 2027, kitty keyboard/graphics, sgr-pixels) keep landing and
    // keep mutating vx.caps from the reader thread, long after
    // enableDetectedFeatures ran and declined to switch those modes on. So
    // over ssh the terminal sat in its default modes while caps claimed
    // otherwise — kitty keyboard was never actually pushed, ever. Raising the
    // timeout only moves the link speed at which that happens.
    //
    // Resolve it on the loop instead. DA1 is last in the probe string and
    // terminals answer in order, so when vaxis's reader flips queries_done
    // every earlier reply is already applied — no window left to miss at any
    // latency, and the enable lands before the next frame (see caps_pending
    // in pollFrame). A terminal that never answers keeps the defaults, which
    // is what the 2ms timeout produced anyway, and with no caps
    // enableDetectedFeatures writes no bytes — the snapshot goldens, where
    // nothing ever answers, do not move. Startup gets 2ms faster, not slower.
    //
    // ponytail: nothing wakes the loop for DA1 alone. In practice the reply
    // burst carries the mode-2048 size report too, which posts a winsize and
    // turns the loop; a terminal idle from boot that lands DA1 between two
    // parks keeps the defaults until the user's first keystroke (decoded
    // legacy, which vaxis handles). Ceiling accepted because nothing in the
    // render path reads the missing caps: pardes writes one codepoint per
    // cell plus an explicit blank spacer under a wide glyph, and vaxis's
    // Cell.width defaults to 1, so gwidth — the only consumer of
    // caps.unicode — is never called. If that ever changes, wake the loop on
    // vx.query_futex from a one-shot thread instead.
    try vx.queryTerminalSend(tty.writer());

    // now threads are fine: start a reader task per pty
    sh.threads_ok = true;
    for (&sh.ptys, 0..) |*slot, id| if (slot.*) |*pt| {
        pt.reader = try io.concurrent(readPty, .{ io, gpa, pt.file, id, sh.gens[id], &loop });
    };
    // ...and the one file watcher. Started here rather than lazily on the
    // first watched pane because the fd already exists and an unwatched
    // inotify instance just parks in read(2) — one thread for the process,
    // however many panes come and go.
    if (sh.inotify_fd >= 0) sh.watch_task = io.concurrent(watchFiles, .{ io, sh.inotify_fd, &loop }) catch null;

    // The core owns the loop ORDER (see Pardes.pump); the outer `while` stays
    // here rather than being `core.run` for one reason: Restore swaps the
    // whole core, and a core cannot replace itself from inside its own frame.
    while (!core.quit) {
        try core.pump(host);
        // Restore builtin: swap in a core rebuilt from the dump; the live
        // shells die with their masters (readers canceled, gens bumped so
        // their late eofs never touch the replay panes)
        if (core.takeRestore()) |rp| blk: {
            const bytes = look.readFile(gpa, rp) catch break :blk;
            defer gpa.free(bytes);
            var o = core.opts;
            o.cols = core.screen_w;
            o.rows = core.screen_h; // pre-size: dump panes never greet
            const nc = pardes.Pardes.initFromDump(allocs.pardes, o, bytes) catch break :blk;
            for (&sh.ptys, 0..) |*slot, pid| if (slot.*) |*pt| {
                pt.reader.cancel(io) catch {};
                _ = libc.close(pt.file.handle);
                slot.* = null;
                sh.gens[pid] +%= 1;
            };
            // the replay core's pane ids mean new things, and the dying core's
            // `watch off` effects go into a queue nobody drains — drop the lot
            // here. The new core emits its own `on`s as it builds its panes.
            for (0..pardes.MAX_PANES) |wid| file_watch.watchPane(
                sh.inotify_fd,
                &sh.watches,
                @intCast(wid),
                null,
                0,
                .{ .text = 0 },
            );
            _ = file_watch.applyThemeEffect(core, gpa, sh.inotify_fd, &sh.watches, 0, false, false);
            var image_iterator = kitty_handles.valueIterator();
            while (image_iterator.next()) |handle| vx.freeImage(tty.writer(), handle.id);
            kitty_handles.clearRetainingCapacity();
            nc.native_images = vx.caps.kitty_graphics;
            core.deinit();
            core = nc;
            sh.core = nc;
            if (comptime pardes.pdf_enabled) {
                // A restored core did not receive the terminal's earlier
                // winsize event. Reapply both the grid and physical cells
                // before its first PDF frame so pointer/pan geometry stays
                // identical to placement.
                updateCoreTerminalSize(core, vx.screen.width, vx.screen.height, vx.screen.width_pix, vx.screen.height_pix);
            }
        }
    }
    // THE FILESYSTEM GOES FIRST, ahead of every deferred teardown below.
    // `loop.stop()` joins a reader parked in `read(2)` on the tty, so it does
    // not return until the next keystroke — and a session that has decided to
    // exit must not spend that wait holding a mount nobody is serving. A
    // client blocked on `<id>/event` when the last pane is deleted through
    // `ctl` then gets ENOTCONN at once instead of hanging until somebody
    // touches the keyboard.
    if (fs) |f| {
        f.deinit();
        fs = null;
        sh.fs = null;
    }
}

/// Everything the terminal shell owns and the core cannot: the ptys, the
/// inotify table, the worker futures, and the one thread allowed to touch
/// `tty.writer()`. This struct IS the `Host.ctx`.
const Shell = struct {
    io: std.Io,
    gpa: std.mem.Allocator,
    lsp_gpa: std.mem.Allocator,
    /// live core; Restore replaces it (see run)
    core: *pardes.Pardes,
    prompt_rcs: *const shell_bin.PromptRcs,
    loop: *Loop,
    vx: *vaxis.Vaxis,
    tty: *vaxis.Tty,
    kitty: *std.AutoHashMap(pardes.ImageCacheKey, vaxis.Image),
    frame: *std.heap.ArenaAllocator,
    /// Where a bracketed paste is assembled. It has to outlive one drain pass:
    /// the burst arrives over as many passes as the terminal takes to write
    /// it, and the markers are the only thing that says where it ends.
    paste_buf: *std.Io.Writer.Allocating,
    inotify_fd: c_int,
    /// acme's control filesystem for this session, or null when `--fs` was not
    /// given (or its mount failed). Owned by `run`, which mounts it before the
    /// first fork and tears it down on every path out.
    fs: ?*fuse.Fs = null,
    ptys: [pardes.MAX_PANES]?Pty = @splat(null),
    /// per-slot spawn generation: a reused pane id ignores the old shell's
    /// late pty_eof (which would otherwise close the NEW pty on that slot)
    gens: [pardes.MAX_PANES]u32 = @splat(0),
    watches: file_watch.Table = @splat(null),
    watch_task: ?std.Io.Future(anyerror!void) = null,
    /// the single in-flight language query (see the lsp method)
    lsp_task: ?std.Io.Future(anyerror!void) = null,
    /// Selection filters may overlap: a second submit supersedes the first in
    /// core without synchronously canceling a possibly slow shell command.
    pipe_tasks: PipeTasks = .{},
    /// false during the pre-loop drain: no worker exists to answer to yet
    threads_ok: bool = false,
    caps_pending: bool = true,
    check_files: bool = false,
    in_paste: bool = false,
    /// the tracks the frame in flight was composed from
    tracks: []const pardes.panel_animation.Track = &.{},

    fn of(ctx: ?*anyopaque) *Shell {
        return @ptrCast(@alignCast(ctx.?));
    }

    fn host(s: *Shell) host_api.Host {
        return .{ .ctx = s, .vtable = if (comptime pardes.isolated) &isolated_vtable else &vtable };
    }

    /// An ISOLATED build (`zig build run-isolated`): the terminal this draws on
    /// and the keys it reads, and nothing else. Every other method stays null,
    /// so the core answers it itself — the embedded source filesystem, the
    /// in-process clipboard, silent ptys. Nothing is forked, nothing on disk is
    /// opened or written, and no clipboard leaves the process. The option is
    /// comptime, so the other vtable is not even built into that binary.
    const isolated_vtable: host_api.Host.VTable = .{
        .pull_wait_input = waitInput,
        .push_present = present,
        .push_post_present = postPresent,
    };

    const vtable: host_api.Host.VTable = .{
        .pull_wait_input = waitInput,
        .push_present = present,
        .push_post_present = postPresent,
        .push_poll_frame = pollFrame,
        .push_spawn = spawn,
        .push_pty_write = ptyWrite,
        .push_pty_resize = ptyResize,
        .pull_tty_taken = ttyTaken,
        .push_write_file = writeFile,
        .push_write_dump = writeDump,
        .push_watch_file = watchFile,
        .push_watch_theme = watchTheme,
        .push_dump_themes = dumpThemes,
        .push_set_clipboard = setClipboard,
        .pull_read_clipboard = readClipboard,
        .push_open_link = openLink,
        .pull_lsp = lsp,
        .pull_pipe = pipe,
        .push_fs_reply = fsReply,
    };

    // ---- input ------------------------------------------------------------

    /// Block for one event, then apply the whole pending batch. Everything
    /// goes through `core.update` rather than `postEvent`: a pty chunk borrows
    /// its bytes for the call, and mixing queued with immediate delivery would
    /// reorder a keystroke against the output it caused.
    fn waitInput(ctx: ?*anyopaque, timeout_ms: u32) void {
        const s = of(ctx);
        var batch: usize = 0;
        if (timeout_ms == 0) {
            // A failed read is a DEAD event source — the reader is gone and no
            // event can ever arrive again, so nothing could set `quit` and the
            // outer `while (!core.quit)` would spin at full speed. End the
            // session, exactly as the pre-vtable `try loop.nextEvent()` did.
            const first = s.loop.nextEvent() catch {
                s.core.quit = true;
                return s.reloadWatched();
            };
            batch = 1;
            if (s.apply(first)) return s.reloadWatched();
        } else {
            // Animating: the frame clock IS the wait, and the `.tick` that
            // spends it is ours to post — `pump` cannot, because only this
            // host knows when a real frame interval has passed. Anything that
            // queues during the short sleep is drained below, in this pass.
            std.Io.sleep(s.io, .fromMilliseconds(timeout_ms), .awake) catch {};
            _ = s.apply(.tick);
            batch = 1;
        }
        // Apply every queued INPUT event, then render ONCE — the gui shell
        // drains SDL's queue the same way. Without this a wheel flick is fifty
        // full render+repaint (and re-highlight) cycles instead of one. A
        // paste in flight keeps draining WITHOUT rendering: a hundred thousand
        // pasted characters are one edit. That cannot spin — the drain still
        // ends the moment the queue runs dry.
        while (s.in_paste or batch < 64) {
            const ev = (s.loop.tryEvent() catch null) orelse break;
            batch += 1;
            if (s.apply(ev)) break;
        }
        s.reloadWatched();
    }

    /// Apply one vaxis event. Returns true when the batch must end here: the
    /// session is over, a pty chunk wants its own frame so progress paints as
    /// it arrives, or a native PDF page crossing needs geometry this render
    /// has not produced yet.
    fn apply(s: *Shell, event: @TypeOf(Command.value)) bool {
        const core = s.core;
        const tz_event = tracy.zone(@src(), "event");
        defer tz_event.end();
        switch (event) {
            .nop => {},
            .tick => core.update(.tick),
            .quit => {
                core.quit = true;
                return true;
            },
            .focus_in => {},
            .focus_out => core.update(.pointer_leave),
            .winsize => |ws| {
                s.vx.resize(s.gpa, s.tty.writer(), ws) catch {};
                if (comptime pardes.pdf_enabled)
                    updateCoreTerminalSize(core, ws.cols, ws.rows, ws.x_pixel, ws.y_pixel)
                else
                    core.update(.{ .resize = .{ .cols = ws.cols, .rows = ws.rows } });
            },
            .pty_read => |pr| {
                core.update(.{ .output = .{ .pane = @intCast(pr.id), .bytes = pr.bytes } });
                s.gpa.free(pr.bytes);
                return true;
            },
            .pty_eof => |e| if (s.gens[e.id] == e.gen) {
                if (s.ptys[e.id]) |*pt| {
                    pt.reader.await(s.io) catch {}; // reader just finished; join it or its future leaks
                    _ = libc.close(pt.file.handle);
                    s.ptys[e.id] = null;
                }
                core.update(.{ .eof = .{ .pane = @intCast(e.id) } });
            },
            .key_press => |key| if (s.in_paste) {
                const bytes = pasteBytes(key);
                const room = max_paste_bytes -| s.paste_buf.written().len;
                s.paste_buf.writer.writeAll(bytes[0..@min(bytes.len, room)]) catch {};
            } else core.update(keyEvent(key)),
            .mouse => |m| {
                const pdf_before = nativePdfWheelTarget(core, m);
                if (mouseEvent(m)) |ev| core.update(ev);
                if (pdf_before) |before| {
                    if (core.panes[before.pane]) |pane| {
                        if (pane.pdfPage()) |page| {
                            if (page != before.page) return true;
                        }
                    }
                }
            },
            .paste => |bytes| {
                core.update(.{ .paste = bytes });
                s.gpa.free(@constCast(bytes));
            },
            .paste_start => {
                s.in_paste = true;
                s.paste_buf.clearRetainingCapacity();
            },
            .paste_end => {
                s.in_paste = false;
                // ONE event for the whole paste — the core borrows the
                // bytes for the call, exactly like the OSC 52 arm above.
                const pasted = s.paste_buf.written();
                if (pasted.len > 0) core.update(.{ .paste = pasted });
                s.paste_buf.clearRetainingCapacity();
            },
            .command => |line| {
                core.update(.{ .command = line });
                s.gpa.free(line);
            },
            // Coalesced on purpose: a burst of writes (a formatter, a build, a
            // `git checkout`) collapses into ONE pass below, so it cannot
            // queue a reload — or an undo entry — per write.
            .files_changed => s.check_files = true,
            // A wake and nothing more. The requests behind it are drained in
            // pollFrame, where the file-watch reload also happens: both want to
            // run once per frame with the whole batch already in, not once per
            // event. So this arm has nothing to do — which is the point, since
            // its only job was ending the blocking wait above.
            .fs_ready => {},
            .lsp_done => |d| {
                core.update(.{ .lsp_resp = .{ .id = d.id, .rows = d.rows } });
                s.lsp_gpa.free(d.rows);
                // the worker is finished; join it so its future does not
                // leak (same contract as pty_eof above)
                if (s.lsp_task) |*t| {
                    t.await(s.io) catch {};
                    s.lsp_task = null;
                }
            },
            .pipe_done => |response_value| {
                var response = response_value;
                core.update(.{ .pipe_resp = .{
                    .id = response.id,
                    .success = response.success,
                    .outputs = response.outputs,
                } });
                response.deinit(s.gpa);
                s.pipe_tasks.finish(s.io, response_value.id);
            },
        }
        return false;
    }

    fn reloadWatched(s: *Shell) void {
        if (!s.check_files) return;
        s.check_files = false;
        if (file_watch.reloadChanged(s.core, s.io, s.gpa, &s.watches))
            s.loop.postEvent(.files_changed) catch {};
    }

    // ---- the frame ---------------------------------------------------------

    fn pollFrame(ctx: ?*anyopaque) void {
        const s = of(ctx);
        // acme's filesystem, answered here for the same reason the file-watch
        // reload is (see reloadWatched): one batch per frame, not one frame per
        // request. First in the pass, so an edit a script just made through
        // `body` is in the surface this frame composes rather than the next.
        if (s.fs) |f| if (fs_service.drain(f, s.core).pending) {
            // The batch hit its cap with requests still pending, and no ack has
            // gone to the poll thread — so nothing else will wake us. Re-arm
            // the loop ourselves. tryPostEvent, not postEvent: this runs on the
            // only thread that drains the queue, so blocking on a full one
            // would deadlock, and a full queue already holds a wake.
            _ = s.loop.tryPostEvent(.fs_ready) catch {};
        };
        // live cwd for tags/look: cheap per-pane lookup, per frame. Whether the
        // pane's tty still belongs to the prompt we forked is NOT polled here —
        // it is a walk through /proc and nothing draws it, so the core pulls it
        // through tty_taken instead, at the Exec that cares.
        for (&s.ptys, 0..) |*slot, id| if (slot.*) |pt| {
            var lbuf: [1024]u8 = undefined;
            if (look.shellCwd(pt.pid, &lbuf)) |cwd| s.core.setCwd(id, cwd);
        };

        // The handshake landed (vaxis's reader flips queries_done on DA1, the
        // last probe answered): put the terminal into the modes the caps now
        // claim, before anything is drawn under them. Polled here rather than
        // done where the replies arrive because that is the reader thread, and
        // this is the only thread allowed to touch the tty writer. The repaint
        // matters as much as the enable: earlier frames were drawn under the
        // pre-handshake caps, and vaxis's shadow grid has to be re-established
        // under the new ones or it keeps skipping cells it thinks are current.
        if (s.caps_pending and s.vx.queries_done.load(.unordered)) {
            s.caps_pending = false;
            s.vx.enableDetectedFeatures(s.tty.writer()) catch {};
            // The core was constructed before the asynchronous handshake.
            // Advertise native pixels only now, after every reply preceding
            // DA1 has updated the capability set.
            s.core.native_images = s.vx.caps.kitty_graphics;
            s.vx.queueRefresh();
        }
    }

    /// The canonical surface -> vaxis, cell for cell, with no interpretation.
    fn present(ctx: ?*anyopaque, canonical: *const pardes.Surface) void {
        const s = of(ctx);
        const vx = s.vx;
        s.tracks = canonical.panelTracks();
        _ = s.frame.reset(.retain_capacity);
        // compose only READS the canonical surface; the mutable pointer is so
        // it can hand it straight back when no panel is mid-transition.
        const surface = panel_compositor.compose(
            s.frame.allocator(),
            @constCast(canonical),
            s.tracks,
            s.core.theme().bg,
        ) catch return;
        const tz_cells = tracy.zone(@src(), "surface->vaxis");
        const win = vx.window();
        paintCells(win, surface.cells, surface.cols, surface.rows);
        tz_cells.end();
        // Pixel attachments (kitty graphics): transmit once per pixel
        // generation, then re-place every frame (placements aren't
        // persistent). The map is keyed by pane lifetime + PDF page + pixel
        // revision, so any number of short visible pages can coexist without
        // aliasing a fixed terminal cache slot.
        for (surface.images[0..surface.nimages]) |maybe| {
            const place = maybe orelse continue;
            // Keep the placement in Surface so the terminal-side cache stays
            // live, but do not pin native pixels over a panel whose cells are
            // currently moving through the TTY grid.
            if (panel_compositor.hidesAttachment(surface.panelTracks(), place.serial)) continue;
            if (comptime pardes.pdf_enabled) {
                if (!kittyImageRepresentable(place)) continue;
            }
            const key = place.cacheKey();
            if (!s.kitty.contains(key) and vx.caps.kitty_graphics) {
                const enc = std.base64.standard.Encoder;
                if (s.gpa.alloc(u8, enc.calcSize(place.rgba.len))) |b64| {
                    defer s.gpa.free(b64);
                    _ = enc.encode(b64, place.rgba);
                    if (vx.transmitPreEncodedImage(s.tty.writer(), b64, @intCast(place.iw), @intCast(place.ih), .rgba) catch null) |handle|
                        s.kitty.put(key, handle) catch vx.freeImage(s.tty.writer(), handle.id);
                } else |_| {}
            }
            if (s.kitty.get(key)) |cached| {
                if (comptime !pardes.pdf_enabled) {
                    const child = win.child(.{ .x_off = place.x, .y_off = place.y, .width = place.w, .height = place.h });
                    cached.draw(child, .{ .scale = .contain }) catch {};
                } else {
                    if (place.native.fit == .contain) {
                        const child = win.child(.{ .x_off = place.x, .y_off = place.y, .width = place.w, .height = place.h });
                        cached.draw(child, .{ .scale = .contain }) catch {};
                    } else if (kittyPlacement(
                        place,
                        vx.screen.width,
                        vx.screen.height,
                        vx.screen.width_pix,
                        vx.screen.height_pix,
                    )) |placement| {
                        const child = win.child(.{
                            .x_off = @as(i17, place.x) + placement.cell_x,
                            .y_off = @as(i17, place.y) + placement.cell_y,
                            .width = placement.cell_cols,
                            .height = placement.cell_rows,
                        });
                        cached.draw(child, placement.options) catch {};
                    }
                }
            }
        }
        // Toggling PETSCII or closing a pane removes its attachment from the
        // Surface. Release the terminal-side image then, not merely when that
        // numeric pane slot happens to be reused.
        while (true) {
            var stale: [pardes.MAX_PANES]pardes.ImageCacheKey = undefined;
            var stale_len: usize = 0;
            var image_iterator = s.kitty.iterator();
            while (image_iterator.next()) |entry| {
                if (surfaceHasKittyKey(surface, entry.key_ptr.*)) continue;
                stale[stale_len] = entry.key_ptr.*;
                stale_len += 1;
                if (stale_len == stale.len) break;
            }
            for (stale[0..stale_len]) |key| if (s.kitty.fetchRemove(key)) |removed|
                vx.freeImage(s.tty.writer(), removed.value.id);
            if (stale_len < stale.len) break;
        }
        if (surface.cursor) |cur| paintCursor(win, cur.x, cur.y, cur.bar);
        const tz_render = tracy.zone(@src(), "vx.render");
        vx.render(s.tty.writer()) catch {};
        tz_render.end();
    }

    fn postPresent(ctx: ?*anyopaque) void {
        const s = of(ctx);
        s.core.acknowledgePanelPresentation(s.tracks);
        tracy.frameMark();
    }

    // ---- pseudo-terminals ---------------------------------------------------

    fn spawn(ctx: ?*anyopaque, pane: u8, cwd: []const u8) void {
        const s = of(ctx);
        // the core reuses pane ids and there is no close effect: a deleted
        // pane's shell lives in its slot until a respawn lands here — reap
        // it (cancel joins the reader; its late eof is ignored by gen)
        if (s.ptys[pane]) |*old| {
            old.reader.cancel(s.io) catch {};
            _ = libc.close(old.file.handle);
            s.ptys[pane] = null;
        }
        s.gens[pane] +%= 1;
        var cwd_buf: [256:0]u8 = undefined;
        var cwd_z: ?[*:0]const u8 = null;
        if (cwd.len > 0 and cwd.len < cwd_buf.len) {
            @memcpy(cwd_buf[0..cwd.len], cwd);
            cwd_buf[cwd.len] = 0;
            cwd_z = @ptrCast(&cwd_buf);
        }
        const child = forkShell(s.core, pane, s.prompt_rcs, s.core.shellBin(), cwd_z, s.core.screen_h, s.core.screen_w, s.fs);
        s.ptys[pane] = .{ .file = child.file, .pid = child.pid, .reader = .{ .any_future = null, .result = {} } };
        // report the pane's starting directory back to the core (tags). The
        // slot needs no occupancy reset: nothing is remembered, and the next
        // Exec asks about the shell that is there now.
        var lbuf: [1024]u8 = undefined;
        if (look.shellCwd(child.pid, &lbuf)) |wd| s.core.setCwd(pane, wd);
        if (s.threads_ok) {
            if (s.ptys[pane]) |*pt| {
                pt.reader = s.io.concurrent(readPty, .{ s.io, s.gpa, pt.file, @as(usize, pane), s.gens[pane], s.loop }) catch pt.reader;
            }
        }
    }

    fn ptyWrite(ctx: ?*anyopaque, pane: u8, bytes: []const u8) void {
        const s = of(ctx);
        if (s.ptys[pane]) |pt| writeFd(pt.file.handle, bytes);
    }

    fn ptyResize(ctx: ?*anyopaque, pane: u8, cols: u16, rows: u16) void {
        const s = of(ctx);
        if (s.ptys[pane]) |pt| {
            const ws: posix.winsize = .{ .row = rows, .col = cols, .xpixel = 0, .ypixel = 0 };
            _ = posix.system.ioctl(pt.file.handle, TIOCSWINSZ, @intFromPtr(&ws));
        }
    }

    /// Is a pane's tty still the prompt we forked? Lazy by construction — it
    /// runs only where the core is about to type a command line, so the /proc
    /// walk costs nothing on an ordinary frame. A pane with no pty of ours (a
    /// document, a slot whose shell already died) is not a terminal a program
    /// can be holding.
    fn ttyTaken(ctx: ?*anyopaque, pane: u8) bool {
        const s = of(ctx);
        const pt = s.ptys[pane] orelse return false;
        return look.ttyTaken(pt.pid, pt.file.handle);
    }

    // ---- the filesystem -----------------------------------------------------

    fn writeFile(ctx: ?*anyopaque, pane: u8, path: []const u8, bytes: []const u8) void {
        const s = of(ctx);
        if (!writeFileBytes(path, bytes)) return;
        // our own write is about to come back as a watch event: restamp from
        // the bytes we just put there so it reads as "no change". Only when
        // this IS the pane's watched file — a `Save <elsewhere>` must not
        // silence a real change to the file the pane has open.
        if (s.core.panes[pane]) |pn| if (pn.file) |f| if (std.mem.eql(u8, f.path, path)) {
            if (s.watches[pane]) |*w| if (w.serial == pn.serial) switch (w.generation) {
                .text => w.generation = .{ .text = std.hash.Wyhash.hash(0, bytes) },
                .pdf => {},
            };
        };
        // ...and say so on the pane's message row. AFTER the write, not beside
        // it: every early return above is a save that did not happen and must
        // not be reported as one.
        var mbuf: [256]u8 = undefined;
        s.core.setMessage(pane, message.stamp(&mbuf, "saved", path));
    }

    fn writeDump(ctx: ?*anyopaque, bytes: []const u8) void {
        const s = of(ctx);
        var pbuf: [1024:0]u8 = undefined;
        const path = pardes.dump.outPath(&pbuf) orelse return;
        if (!writeFileBytes(path, bytes)) return;
        s.core.setLastDump(path);
    }

    fn watchFile(ctx: ?*anyopaque, pane: u8, _: []const u8, on: bool) void {
        const s = of(ctx);
        if (file_watch.applyEffect(s.core, s.io, s.gpa, s.inotify_fd, &s.watches, pane, on))
            s.loop.postEvent(.files_changed) catch {};
    }

    fn watchTheme(ctx: ?*anyopaque, generation: u32, on: bool) void {
        const s = of(ctx);
        if (file_watch.applyThemeEffect(s.core, s.gpa, s.inotify_fd, &s.watches, generation, on, s.threads_ok))
            s.loop.postEvent(.files_changed) catch {};
    }

    /// The core's answer to one filesystem request, handed straight back to the
    /// transport that is holding it. `bytes` was resolved by `pardes.fsPayload`
    /// inside `perform` and is borrowed only for this call — a body read is a
    /// window onto the pane's live text, so there is nothing to copy and
    /// nothing to free. `.again` needs no special case here: `Fs.reply` reads
    /// the status and re-parks the request itself.
    fn fsReply(ctx: ?*anyopaque, reply: *const pardes.acmefs.Reply, bytes: []const u8) void {
        const s = of(ctx);
        if (s.fs) |f| f.reply(reply, bytes);
    }

    fn dumpThemes(ctx: ?*anyopaque, pane: u8) void {
        const s = of(ctx);
        const config_dir = s.core.opts.config_dir orelse return;
        const out_dir = user_config.dumpThemes(s.io, s.gpa, config_dir, pardes.themes) catch |err| {
            s.core.reportError(pane, "dump themes", err);
            return;
        };
        defer s.gpa.free(out_dir);
        var mbuf: [256]u8 = undefined;
        s.core.setMessage(pane, message.stamp(&mbuf, "dumped themes", out_dir));
    }

    // ---- the desktop --------------------------------------------------------

    /// mirror the core's yank register out via OSC 52
    fn setClipboard(ctx: ?*anyopaque, text: []const u8) void {
        const s = of(ctx);
        if (text.len == 0) return;
        s.vx.copyToSystemClipboard(s.tty.writer(), text, s.gpa) catch {};
    }

    /// ...and the other direction, OSC 52 read. The answer arrives on vaxis's
    /// reader thread as an ordinary `.paste` event and reaches the core
    /// through the same path an outer bracketed paste does — this request is
    /// the only wiring it needs. "The answer arrives" is the optimistic
    /// reading: a clipboard READ is an exfiltration primitive and terminals
    /// treat it as one (ghostty prompts by default, xterm ships it off, a
    /// multiplexer or ssh link may eat it), and a refusal looks exactly like
    /// silence. So the core's pending request is dropped by the next keystroke
    /// rather than pasting minutes late, and `SPC p` in a locked-down terminal
    /// honestly does nothing.
    fn readClipboard(ctx: ?*anyopaque) void {
        const s = of(ctx);
        s.vx.requestSystemClipboard(s.tty.writer()) catch {};
    }

    fn openLink(_: ?*anyopaque, url: []const u8) void {
        look.openLink(url); // desktop browser
    }

    // ---- work that must leave the loop --------------------------------------

    fn lsp(ctx: ?*anyopaque, req: host_api.LspRequest) void {
        const s = of(ctx);
        if (!s.threads_ok) return; // pre-loop drain: nothing to answer to yet
        const pane = s.core.panes[req.pane] orelse return;
        // a pane with no file still asks `status` (it is about the backend,
        // not the buffer): empty path and source, root from the pane's cwd
        const f = pane.file;
        const a = s.lsp_gpa;
        const job = a.create(LspJob) catch return;
        job.* = .{
            .id = req.id,
            .kind = req.kind,
            .offset = req.offset,
            .path = a.dupe(u8, if (f) |ff| ff.path else "") catch {
                a.destroy(job);
                return;
            },
            .source = a.dupeZ(u8, if (f) |ff| ff.content else "") catch {
                a.free(job.path);
                a.destroy(job);
                return;
            },
            .arg = a.dupe(u8, req.arg) catch {
                a.free(job.path);
                a.free(job.source);
                a.destroy(job);
                return;
            },
            .root = a.dupe(u8, if (f) |ff| (std.fs.path.dirname(ff.path) orelse "/") else pane.cwdSlice()) catch {
                a.free(job.path);
                a.free(job.source);
                a.free(job.arg);
                a.destroy(job);
                return;
            },
        };
        // ponytail: ONE query in flight, so one future slot. Replacing it
        // cancels-then-joins the previous worker, which for a backend that
        // ignores cancellation means waiting out a query the user already
        // abandoned. Queries are milliseconds; make this a real pool the day a
        // backend takes long enough to notice.
        if (s.lsp_task) |*old| {
            old.cancel(s.io) catch {};
            s.lsp_task = null;
        }
        s.lsp_task = s.io.concurrent(lspWorker, .{ a, job, s.loop }) catch {
            job.free(a);
            return;
        };
    }

    fn pipe(ctx: ?*anyopaque, id: u32) void {
        const s = of(ctx);
        if (!s.threads_ok) return;
        if (s.pipe_tasks.full()) {
            s.core.update(.{ .pipe_resp = .{ .id = id, .success = false, .outputs = &.{} } });
            return;
        }
        const view = s.core.pipeRequest(id) orelse return;
        const job = selection_pipe.Job.copy(s.gpa, view) catch return;
        const future = s.io.concurrent(pipeWorker, .{ s.io, s.gpa, job, s.loop }) catch {
            job.deinit(s.gpa);
            return;
        };
        s.pipe_tasks.add(.{ .id = id, .future = future });
    }
};

/// Answer a language query off the event loop and post the rows back. This is
/// the whole async execution model: the same shape as readPty — do the slow
/// thing on a worker, hand the result to the loop as an event, let the core
/// stay a state machine that never blocks.
fn lspWorker(allocator: std.mem.Allocator, job: *LspJob, loop: *Loop) anyerror!void {
    defer job.free(allocator);
    var arena: std.heap.ArenaAllocator = .init(allocator);
    defer arena.deinit();
    // The shell owns the result buffer; the backend only ever writes to it.
    var out: std.Io.Writer.Allocating = .init(allocator);
    defer out.deinit();
    pardes.lsp.query(allocator, arena.allocator(), .{
        .kind = job.kind,
        .path = job.path,
        .source = job.source,
        .offset = job.offset,
        .arg = job.arg,
        .root = job.root,
    }, &out.writer);
    const rows = allocator.dupe(u8, out.written()) catch return;
    loop.postEvent(.{ .lsp_done = .{ .id = job.id, .rows = rows } }) catch allocator.free(rows);
    return;
}

fn pipeWorker(
    io: std.Io,
    gpa: std.mem.Allocator,
    job: *selection_pipe.Job,
    loop: *Loop,
) anyerror!void {
    defer job.deinit(gpa);
    var response = selection_pipe.runJob(gpa, io, job);
    loop.postEvent(.{ .pipe_done = response }) catch response.deinit(gpa);
}

/// Block on the inotify fd and wake the loop. Deliberately does NOT parse the
/// events: the loop re-reads every watched pane anyway, so the only thing an
/// event carries that we need is THAT something happened, and parsing would
/// mean sharing the watch table with the thread that mutates it. Same shape as
/// readPty — block off the loop, hand the loop an event, keep the core a state
/// machine that never blocks. Going through std.Io.File rather than a raw
/// read(2) is what lets the teardown `cancel` interrupt it.
///
/// ponytail: churn in a watched directory that never touches the watched file
/// still costs a wake and a re-read per event. The ceiling is one directory
/// per open file pane; filter by basename here if it ever shows up in a
/// profile.
fn watchFiles(io: std.Io, fd: c_int, loop: *Loop) anyerror!void {
    const file: std.Io.File = .{ .handle = fd, .flags = .{ .nonblocking = false } };
    var read_buf: [4096]u8 = undefined;
    var reader = file.readerStreaming(io, &read_buf);
    while (true) {
        var buf: [4096]u8 = undefined;
        var vec = [_][]u8{&buf};
        const n = reader.interface.readVec(&vec) catch break;
        if (n == 0) break;
        loop.postEvent(.files_changed) catch break;
    }
}

/// Block on the nested-instance socket and hand the loop each command line a
/// pardes started inside this one sends. Same shape as winchWatch: a plain
/// detached thread around a call that never returns, posting into the vaxis
/// loop from ordinary thread context.
///
/// Nothing here is woken by teardown — close(2) does NOT release a thread
/// parked in accept4 on linux — so this dies with the process, exactly as
/// winchWatch dies inside sigwait. The window that leaves is one connection
/// accepted between the last drain and process exit posting into a queue whose
/// owner has returned; same shape and same bound as every other detached
/// worker here, and a self-pipe to close it would be more machinery than the
/// window is worth.
fn lookServer(gpa: std.mem.Allocator, fd: c_int, loop: *Loop) void {
    var buf: [nested.max_line]u8 = undefined;
    while (nested.acceptLine(fd, &buf)) |line| {
        const owned = gpa.dupe(u8, line) catch continue;
        loop.postEvent(.{ .command = owned }) catch {
            gpa.free(owned);
            break;
        };
    }
}

/// Consume SIGWINCH synchronously (it is blocked in every thread) and post
/// the new size as a winsize event from normal thread context — the one place
/// vaxis's Io-backed queue is safe to touch on a resize.
fn winchWatch(loop: *Loop, vx: *vaxis.Vaxis, tty: *vaxis.Tty) void {
    var set = posix.sigemptyset();
    posix.sigaddset(&set, posix.SIG.WINCH);
    while (true) {
        var sig: c_int = 0;
        if (libc.sigwait(&set, &sig) != 0) continue;
        if (vx.state.in_band_resize) continue; // terminal reports via CSI 48
        const ws = tty.getWinsize() catch continue;
        loop.postEvent(.{ .winsize = ws }) catch {};
    }
}

/// The /dev/fuse poller's wake, and deliberately nothing else — the thread that
/// calls this has no business in the core, so all it does is end the blocking
/// `nextEvent`. tryPostEvent rather than postEvent for the same reason readPty's
/// final post uses it: this can fire after the loop has already been left, and a
/// blocking push into a full queue nobody is draining would never return.
fn wakeFs(ctx: ?*anyopaque) void {
    const loop: *Loop = @ptrCast(@alignCast(ctx.?));
    _ = loop.tryPostEvent(.fs_ready) catch {};
}

/// `core` is null in an `--attach` frontend, which forks the pane shells for a
/// core that is in another process entirely. The two things it is used for are
/// both messages BACK to that core — which pane serial the child's environment
/// should name, and which binary was actually executed — and neither has a
/// place on the detached wire (see wire.zig): a detached session's `--fs`
/// stays in the session, and `acknowledgeShell` has no `ClientTag`. So both
/// are skipped rather than faked, and the pane tag in a detached session
/// simply does not name its shell.
fn forkShell(core: ?*pardes.Pardes, pane: usize, prompt_rcs: *const shell_bin.PromptRcs, bin: []const u8, cwd: ?[*:0]const u8, rows: u16, cols: u16, fs: ?*const fuse.Fs) struct { file: std.Io.File, pid: posix.pid_t } {
    var master: c_int = undefined;
    // resolved BEFORE the fork, into this frame, which the child inherits:
    // nothing between fork and exec may allocate, and a PATH search would
    var path_buf: [std.fs.max_path_bytes]u8 = undefined;
    const spawn = shell_bin.resolve(bin, &path_buf, prompt_rcs);
    // ...and so is the pane's own address on the control filesystem, for a
    // second reason on top of that one: acme puts `winid` in the child, which
    // is safe there only because rfork(RFENVG) has just given it a private
    // environment group. See fs_service.exportPaneEnv.
    fs_service.exportPaneEnv(fs, if (core) |c| (if (c.panes[pane]) |pn| pn.serial else 0) else 0);
    const ws = posix.winsize{ .row = rows, .col = cols, .xpixel = 0, .ypixel = 0 };
    const pid = forkpty(&master, null, null, &ws);
    if (pid == 0) {
        // the blocked-SIGWINCH mask survives fork AND exec — unblock it or
        // bash/vim in the pane would never see resizes (sigprocmask is
        // async-signal-safe)
        var set = posix.sigemptyset();
        posix.sigaddset(&set, posix.SIG.WINCH);
        posix.sigprocmask(posix.SIG.UNBLOCK, &set, null);
        if (cwd) |c| _ = chdir(c);
        _ = execv(spawn.path, &spawn.argv);
        _exit(127);
    }
    if (pid > 0) if (core) |c| c.acknowledgeShell(pane, std.mem.span(spawn.path), spawn.argv[1] != null);
    return .{ .file = .{ .handle = master, .flags = .{ .nonblocking = false } }, .pid = pid };
}

fn readPty(io: std.Io, gpa: std.mem.Allocator, pty: std.Io.File, id: usize, gen: u32, loop: *Loop) anyerror!void {
    var read_buf: [0x10000]u8 = undefined;
    var reader = pty.readerStreaming(io, &read_buf);
    while (true) {
        var buf: [0x10000]u8 = undefined;
        var vec = [_][]u8{&buf};
        const n = reader.interface.readVec(&vec) catch break;
        if (n == 0) break;
        const bytes = try gpa.dupe(u8, buf[0..n]);
        loop.postEvent(.{ .pty_read = .{ .id = id, .bytes = bytes } }) catch {
            gpa.free(bytes);
            break;
        };
    }
    // non-blocking: a teardown cancel only unblocks one wait, so a blocking
    // post into a full queue here could hang the exit
    _ = loop.tryPostEvent(.{ .pty_eof = .{ .id = id, .gen = gen } }) catch {};
}

/// The effective codepoint the way vaxis Key.matches sees it: a single-char
/// text wins (shift resolved by the terminal), else the shifted codepoint.
fn effCp(key: vaxis.Key) u21 {
    if (key.text) |t| {
        const view = std.unicode.Utf8View.init(t) catch return key.codepoint;
        var it = view.iterator();
        if (it.nextCodepoint()) |c| {
            if (it.nextCodepoint() == null) return c;
        }
    }
    return key.shifted_codepoint orelse key.codepoint;
}

/// vaxis functional-key codepoints -> core Key constants (ASCII ones already
/// coincide: enter/tab/escape/backspace pass through).
fn mapKey(cp: u21) u21 {
    return switch (cp) {
        vaxis.Key.up => pardes.Key.up,
        vaxis.Key.down => pardes.Key.down,
        vaxis.Key.left => pardes.Key.left,
        vaxis.Key.right => pardes.Key.right,
        vaxis.Key.home => pardes.Key.home,
        vaxis.Key.end => pardes.Key.end,
        vaxis.Key.page_up => pardes.Key.page_up,
        vaxis.Key.page_down => pardes.Key.page_down,
        vaxis.Key.delete => pardes.Key.delete,
        else => cp,
    };
}

/// 4 MiB ceiling, past which the tail is dropped rather than grown into. A
/// paste that large is a mis-click on a file, not an edit, and the core would
/// have to hold the whole of it as one undo entry. File scope rather than a
/// `Shell` decl because the `--attach` frontend accumulates the same bursts
/// against the same ceiling, and wire.zig cites this name as THE cap.
const max_paste_bytes: usize = 4 << 20;

/// One key press between the bracketed-paste markers, as the bytes it means.
/// A key in there is DATA, never a command, and the two callers — the
/// in-process host and the `--attach` frontend — must agree exactly, because
/// what they produce is compared against what a terminal's own paste would
/// have delivered.
fn pasteBytes(key: vaxis.Key) []const u8 {
    const text = key.text orelse "";
    if (text.len > 0) return text;
    const cp = mapKey(effCp(key));
    if (cp == pardes.Key.tab) return "\t";
    // vaxis gives control bytes no text at all: a line break inside a paste
    // reaches the ground parser as a bare CR (-> Key.enter) or, from a
    // terminal that does not translate them, a bare LF — which that parser
    // reports as ctrl+j. Nothing in here is a real keypress, so both of them
    // are just a newline.
    if (cp == pardes.Key.enter or (key.mods.ctrl and cp == 'j')) return "\n";
    // arrows, F-keys, a stray escape: noise a paste has no business carrying,
    // dropped rather than smuggled in.
    return "";
}

/// ...and one ordinary key press as the core's event. Shared for the reason
/// above: a keystroke must mean the same thing whether the core is in this
/// process or on the other end of a socket.
fn keyEvent(key: vaxis.Key) pardes.Event {
    return .{ .key = .{
        .cp = mapKey(effCp(key)),
        .text = key.text orelse "",
        .ctrl = key.mods.ctrl,
        .alt = key.mods.alt,
        .shift = key.mods.shift,
    } };
}

/// ...and one mouse report. Null for a button this vocabulary has no name for
/// (vaxis reports more of them than the core has), which is a report to drop
/// rather than a press to invent.
fn mouseEvent(m: vaxis.Mouse) ?pardes.Event {
    const button: pardes.Mouse.Button = switch (m.button) {
        .left => .left,
        .middle => .middle,
        .right => .right,
        .wheel_up => .wheel_up,
        .wheel_down => .wheel_down,
        .wheel_left => .wheel_left,
        .wheel_right => .wheel_right,
        .none => .none, // button-less motion: hover tracking
        else => return null,
    };
    return .{ .mouse = .{
        .button = button,
        .kind = switch (m.type) {
            .press => .press,
            .release => .release,
            .motion => .motion,
            .drag => .drag,
        },
        .col = @intCast(m.col),
        .row = @intCast(m.row),
        .ctrl = m.mods.ctrl,
    } };
}

/// THE cell walk: canonical cells -> vaxis, cell for cell, with no
/// interpretation. One walk and two callers, because a `frame` off the
/// detached wire IS a `Surface`'s cells — wire.zig carries the grid and
/// deliberately does not carry the two halves `Shell.present` adds around this
/// (the kitty attachments, which have no encoding, and the panel transition,
/// which the session composes before it sends). A second walk here would be
/// two renderers for one canonical interface, and the interface is the thing
/// this editor is.
fn paintCells(win: vaxis.Window, cells: []const pardes.Cell, cols: u16, rows: u16) void {
    win.clear();
    var y: u16 = 0;
    while (y < rows) : (y += 1) {
        var x: u16 = 0;
        while (x < cols) : (x += 1) {
            const cell = &cells[@as(usize, y) * cols + x];
            if (cell.default) continue;
            win.writeCell(x, y, .{
                .char = .{ .grapheme = cell.grapheme() },
                .style = vaxisStyle(cell.style),
            });
        }
    }
}

fn paintCursor(win: vaxis.Window, x: u16, y: u16, bar: bool) void {
    win.showCursor(x, y);
    // insert = beam, everything else = the terminal's default shape
    win.setCursorShape(if (bar) .beam else .default);
}

/// Create-or-truncate `path` and put `bytes` there. The half of `write_file`
/// that is nothing but the filesystem, so that the frontend which has a disk
/// and no core and the host which has both write a file the same way.
/// Everything else in `Shell.writeFile` — the watch restamp, the "saved"
/// message — is the CORE's memory of the write and stays with whoever owns
/// one. False is a save that did not happen, which no caller may report as
/// one.
fn writeFileBytes(path: []const u8, bytes: []const u8) bool {
    var pathbuf: [4096:0]u8 = undefined;
    if (path.len >= pathbuf.len) return false;
    @memcpy(pathbuf[0..path.len], path);
    pathbuf[path.len] = 0;
    const fd = libc.open(pathbuf[0..path.len :0], .{ .ACCMODE = .WRONLY, .CREAT = true, .TRUNC = true }, @as(libc.mode_t, 0o644));
    if (fd < 0) return false;
    writeFd(fd, bytes);
    _ = libc.close(fd);
    return true;
}

fn vaxisStyle(s: pardes.CellStyle) vaxis.Style {
    return .{
        .fg = vaxisColor(s.fg),
        .bg = vaxisColor(s.bg),
        .bold = s.bold,
        .dim = s.dim,
        .italic = s.italic,
        .blink = s.blink,
        .reverse = s.reverse,
        .invisible = s.invisible,
        .strikethrough = s.strikethrough,
        .ul_style = switch (s.ul) {
            .off => .off,
            .single => .single,
            .double => .double,
            .curly => .curly,
            .dotted => .dotted,
            .dashed => .dashed,
        },
    };
}

fn vaxisColor(c: pardes.Color) vaxis.Color {
    return switch (c) {
        .default => .default,
        .index => |i| .{ .index = i },
        .rgb => |rgb| .{ .rgb = rgb },
    };
}

fn writeFd(fd: c_int, data: []const u8) void {
    var off: usize = 0;
    while (off < data.len) {
        const n = libc.write(fd, data[off..].ptr, data.len - off);
        if (n < 0) {
            if (libc.errno(n) == .INTR) continue;
            return;
        }
        off += @intCast(n);
    }
}

// ---------------------------------------------------------------------------
// --attach: a terminal, a socket, and no core
// ---------------------------------------------------------------------------

/// Why an `--attach` frontend stopped, and where its exit status comes from.
/// A VALUE rather than a message printed where it is discovered: at that point
/// the alt screen is still up and everything written to it is erased by the
/// restore a moment later. `run` registers `report` BEFORE it opens the
/// terminal, so LIFO runs it last — on a cooked main screen, which is the one
/// screen a person would go looking at.
const AttachEnd = union(enum) {
    /// not an `--attach` run at all, or the session said `quit`: exit 0
    none,
    /// `--attach=<name>` and nothing is listening under it
    no_session: []const u8,
    /// bare `--attach` with no session to mean...
    nothing_detached,
    /// ...or more than one, which is a choice and not ours to make
    ambiguous: usize,
    /// the session hung up on the connect and said why
    refused: wire.Refusal,
    /// the link died, or the stream stopped making sense
    lost: anyerror,
    /// the connect landed and the session hung up before the hello was
    /// answered: a refusal whose reason raced the close (see `attachSession`)
    rejected,

    /// The nonzero exit lives HERE for the same reason the message does: every
    /// teardown this process owes is a defer registered after this one, so by
    /// the time this runs they have all run and there is nothing left to skip.
    fn report(e: AttachEnd, io: std.Io) void {
        var buf: [512]u8 = undefined;
        const text: []const u8 = switch (e) {
            .none => return,
            .no_session => |name| std.fmt.bufPrint(
                &buf,
                "pardes: no detached session called '{s}' (start one with `pardes --detach={s}`)\n",
                .{ name, name },
            ) catch "pardes: no detached session under that name\n",
            .nothing_detached => "pardes: no detached session is running (start one with `pardes --detach`)\n",
            .ambiguous => |n| std.fmt.bufPrint(
                &buf,
                "pardes: {d} detached sessions are running; say which with --attach=<name>\n",
                .{n},
            ) catch "pardes: several detached sessions are running; say which with --attach=<name>\n",
            .refused => |why| switch (why) {
                .version => "pardes: that session speaks a different wire version — it is another build of pardes\n",
                .full => "pardes: that session already has every frontend slot taken\n",
                .quitting => "pardes: that session is ending\n",
            },
            .lost => |err| std.fmt.bufPrint(&buf, "pardes: detached session lost: {t}\n", .{err}) catch
                "pardes: detached session lost\n",
            .rejected => "pardes: that session hung up on the connect — every frontend slot is taken (32), or it is shutting down. `PARDES_LOG=1` on the session names which\n",
        };
        std.Io.File.stderr().writeStreamingAll(io, text) catch {};
        std.process.exit(1);
    }
};

/// The session `--attach` meant. `--attach=<name>` is the name it says; bare
/// `--attach` is THE session, because bare `--detach` names itself by its own
/// pid and nobody can be expected to read a pid out of `$XDG_RUNTIME_DIR`.
/// With exactly one session listening that is the one meant; with none or
/// several this says which case it is instead of picking one. Null means "do
/// not open a terminal", with `end` already saying why.
///
/// Both the directory and the filename convention come off ONE probe through
/// `server.sessionPath`, rather than being re-derived here, for the reason that
/// function exists at all: the side that binds and the side that looks must not
/// be able to disagree about where a session lives.
fn sessionName(buf: *[detached_server.path_max]u8, requested: []const u8, end: *AttachEnd) ?[]const u8 {
    if (requested.len != 0) {
        // A name is taken at its word — the connect in `attachSession` is the
        // authority on whether anything is listening — except for the one case
        // that is worth catching before a terminal is opened at all: a typo,
        // where there is no socket file of that name whatsoever. Getting that
        // wrong is the common failure, and the alternative is a full-screen
        // alt-screen flash on the way to a one-line message.
        var one_buf: [detached_server.path_max]u8 = undefined;
        const one = detached_server.sessionPath(&one_buf, requested) orelse {
            end.* = .{ .no_session = requested };
            return null;
        };
        const F_OK: c_int = 0;
        if (libc.access(one, F_OK) != 0) {
            end.* = .{ .no_session = requested };
            return null;
        }
        return requested;
    }
    const probe_name = "0";
    var probe_buf: [detached_server.path_max]u8 = undefined;
    const probe = detached_server.sessionPath(&probe_buf, probe_name) orelse {
        end.* = .nothing_detached;
        return null;
    };
    const base = std.fs.path.basename(probe);
    const cut = std.mem.lastIndexOf(u8, base, probe_name).?;
    const prefix = base[0..cut];
    const suffix = base[cut + probe_name.len ..];
    var dir_buf: [detached_server.path_max:0]u8 = undefined;
    const dir = std.fs.path.dirname(probe) orelse "";
    if (dir.len == 0 or dir.len >= dir_buf.len) {
        end.* = .nothing_detached;
        return null;
    }
    @memcpy(dir_buf[0..dir.len], dir);
    dir_buf[dir.len] = 0;
    const d = libc.opendir(dir_buf[0..dir.len :0]) orelse {
        end.* = .nothing_detached;
        return null;
    };
    defer _ = libc.closedir(d);
    var found: usize = 0;
    var len: usize = 0;
    while (libc.readdir(d)) |ent| {
        const entry = std.mem.sliceTo(&ent.name, 0);
        if (entry.len <= prefix.len + suffix.len) continue;
        if (!std.mem.startsWith(u8, entry, prefix) or !std.mem.endsWith(u8, entry, suffix)) continue;
        const name = entry[prefix.len .. entry.len - suffix.len];
        if (name.len > buf.len) continue;
        found += 1;
        @memcpy(buf[0..name.len], name);
        len = name.len;
    }
    // A socket file whose session is gone still counts here: the sweep that
    // unlinks corpses runs when the NEXT session binds (server.zig `sweep`),
    // and probing every candidate with a connect would put a phantom frontend
    // into a live session's slot table just to count it. One stale file
    // therefore fails at `open` with "no such session", which is the truth.
    if (found != 1) {
        end.* = if (found == 0) .nothing_detached else .{ .ambiguous = found };
        return null;
    }
    return buf[0..len];
}

/// `--attach[=<name>]`: the frontend half of a detached session. This process
/// owns a terminal and a socket, and the `Pardes` is in the session process
/// (src/detached/). The whole job is client.zig's two sentences — send the
/// input it collects, draw the frames it is sent — plus the real host work a
/// daemon has no way to do and asks a frontend for: fork a shell on a real tty,
/// put bytes on a real disk, reach a real clipboard.
///
/// It is NOT a `Shell`, and the difference is not size. `Shell` IS the
/// `Host.ctx` of a core in THIS process, and its methods reach into that core
/// on nearly every line — a pane's message row, a watch generation taken off
/// the pane's live text, `acknowledgeShell`, `setCwd`. With no core those are
/// not cheaper versions of the same work, they are absent, and each one is
/// named below where it goes missing. What the two do genuinely share is
/// shared: the cell walk, the key and mouse vocabularies, the paste ceiling,
/// `forkShell`, `readPty`, `writeFileBytes`.
const Attach = struct {
    io: std.Io,
    gpa: std.mem.Allocator,
    client: *detached_client.Client,
    loop: *Loop,
    vx: *vaxis.Vaxis,
    tty: *vaxis.Tty,
    prompt_rcs: *const shell_bin.PromptRcs,
    paste_buf: *std.Io.Writer.Allocating,
    /// Where the config directory came from is main.zig's answer, carried in
    /// `Options` — the only field of it this frontend reads, since every other
    /// one describes a core that is elsewhere.
    config_dir: ?[]const u8,
    ptys: [pardes.MAX_PANES]?Pty = @splat(null),
    gens: [pardes.MAX_PANES]u32 = @splat(0),
    inotify_fd: c_int,
    watches: file_watch.Table = @splat(null),
    /// What each watched slot is watching. `file_watch.Table` remembers the
    /// directory mark and the accepted generation but not the pathname — the
    /// in-process host reads that back off the pane, and this one has no panes,
    /// so the path the `watch_file` message carried is kept here.
    watch_paths: [pardes.MAX_PANES]?[]u8 = @splat(null),
    watch_task: ?std.Io.Future(anyerror!void) = null,
    caps_pending: bool = true,
    check_files: bool = false,
    in_paste: bool = false,
    /// A frame landed. Painted once at the end of the round rather than where
    /// it arrives: several can be decoded out of one poll and only the last of
    /// them is on the screen.
    dirty: bool = false,

    /// One event onto the wire. Non-null ends this frontend.
    fn send(a: *Attach, ev: pardes.Event) ?AttachEnd {
        a.client.send(.{ .event = ev }) catch |err| switch (err) {
            // A message this protocol cannot carry, which is not a link that
            // has died: `putSlice32` refuses past `wire.max_payload` (16 MiB),
            // and the one event that can reach it is a `file_changed` for a
            // watched file between that and `look.readFile`'s own 256 MiB cap.
            // Dropping it costs one reload; treating it as a hangup would cost
            // the session.
            error.Overlong, error.NoSpace => return null,
            else => return .{ .lost = err },
        };
        return null;
    }

    /// One decoded message from the session. Every arm of `wire.ServerMsg` is
    /// named and none of them is a catch-all: a session goes on asking for what
    /// it is not given, and the two this frontend genuinely cannot do say so
    /// where they are handled rather than vanishing into an `else`.
    fn handle(a: *Attach, msg: wire.ServerMsg) ?AttachEnd {
        switch (msg) {
            // Already applied to the client's slot and geometry; the full frame
            // the session promises a fresh attach is the next thing to arrive.
            .welcome => {},
            .refuse => |why| return .{ .refused = why },
            // Applied too — `grid` and `cursor` are current by the time this
            // returns, so all that is left is to say the screen moved.
            .frame => a.dirty = true,
            .quit => return .none,

            .spawn => |v| a.spawn(v.pane, v.cwd),
            .pty_write => |v| if (a.ptys[v.pane]) |pt| writeFd(pt.file.handle, v.bytes),
            .pty_resize => |v| if (a.ptys[v.pane]) |pt| {
                const ws: posix.winsize = .{ .row = v.rows, .col = v.cols, .xpixel = 0, .ypixel = 0 };
                _ = posix.system.ioctl(pt.file.handle, TIOCSWINSZ, @intFromPtr(&ws));
            },
            .write_file => |v| a.writeFile(v.pane, v.path, v.bytes),
            // The dump lands on this frontend's disk. WHERE it landed is the
            // core's own memory of it (`setLastDump`), and there is no
            // `ClientTag` to carry a path back, so a detached `Dump` writes the
            // file and the session cannot then name it.
            .write_dump => |bytes| {
                var pbuf: [1024:0]u8 = undefined;
                const path = pardes.dump.outPath(&pbuf) orelse return null;
                _ = writeFileBytes(path, bytes);
            },
            .watch_file => |v| a.watchFile(v.pane, v.path, v.on),
            // NOT DOABLE FROM HERE, and it is the wire's shape rather than an
            // omission: this message carries `{generation, on}`, the theme
            // file's PATH lives in the core (`themeFileRequest`), and there is
            // no message that would carry the reloaded bytes back. So a
            // detached session does not live-reload a theme file. Named
            // anyway, because the alternative is an `else` that would also
            // swallow the next arm somebody adds to the protocol.
            .watch_theme => {},
            .dump_themes => a.dumpThemes(),
            .set_clipboard => |text| if (text.len != 0) {
                a.vx.copyToSystemClipboard(a.tty.writer(), text, a.gpa) catch {};
            },
            // The answer is not a reply message: it comes back as an ordinary
            // `Event.paste`, which is the same asynchronous shape the
            // in-process host has (see `Shell.readClipboard`), and it arrives
            // through the `.paste` arm of `apply` like any other.
            .read_clipboard => a.vx.requestSystemClipboard(a.tty.writer()) catch {},
            .open_link => |url| look.openLink(url),
        }
        return null;
    }

    /// One vaxis event, translated onto the wire. Non-null ends the loop.
    fn apply(a: *Attach, event: @TypeOf(Command.value)) ?AttachEnd {
        switch (event) {
            // Nothing posts these here, and each absence has a reason. `tick`
            // is `Shell.waitInput`'s animation clock, and an animation runs
            // where the core is — the session sleeps on its own frame interval
            // (server.zig `nap`) and the frames simply arrive. `fs_ready`
            // belongs to `--fs`, which wire.zig keeps in the detached process.
            // `lsp_done`/`pipe_done` answer work the core dispatches, and it
            // dispatches it there.
            .nop, .tick, .fs_ready, .lsp_done, .pipe_done => {},
            .quit => return .none,
            .focus_in => {},
            .focus_out => return a.send(.pointer_leave),
            .winsize => |ws| {
                a.vx.resize(a.gpa, a.tty.writer(), ws) catch {};
                // Repaint from the frame already in hand: vaxis has just thrown
                // its shadow grid away, and the SESSION grid may not move at
                // all — it is the smallest common one and another frontend may
                // be the small one (client.zig GEOMETRY).
                a.dirty = true;
                a.client.resize(ws.cols, ws.rows) catch |err| return .{ .lost = err };
            },
            .pty_read => |pr| {
                defer a.gpa.free(pr.bytes);
                return a.send(.{ .output = .{ .pane = @intCast(pr.id), .bytes = pr.bytes } });
            },
            .pty_eof => |e| if (a.gens[e.id] == e.gen) {
                if (a.ptys[e.id]) |*pt| {
                    pt.reader.await(a.io) catch {}; // reader just finished; join it or its future leaks
                    _ = libc.close(pt.file.handle);
                    a.ptys[e.id] = null;
                }
                return a.send(.{ .eof = .{ .pane = @intCast(e.id) } });
            },
            .key_press => |key| if (a.in_paste) {
                const bytes = pasteBytes(key);
                const room = max_paste_bytes -| a.paste_buf.written().len;
                a.paste_buf.writer.writeAll(bytes[0..@min(bytes.len, room)]) catch {};
            } else return a.send(keyEvent(key)),
            .mouse => |m| if (mouseEvent(m)) |ev| return a.send(ev),
            .paste => |bytes| {
                defer a.gpa.free(@constCast(bytes));
                return a.send(.{ .paste = bytes });
            },
            .paste_start => {
                a.in_paste = true;
                a.paste_buf.clearRetainingCapacity();
            },
            .paste_end => {
                a.in_paste = false;
                defer a.paste_buf.clearRetainingCapacity();
                // ONE message for the whole paste, exactly as the in-process
                // host makes it one `update`.
                const pasted = a.paste_buf.written();
                if (pasted.len > 0) return a.send(.{ .paste = pasted });
            },
            .command => |line| {
                defer a.gpa.free(line);
                return a.send(.{ .command = line });
            },
            .files_changed => a.check_files = true,
        }
        return null;
    }

    fn spawn(a: *Attach, pane: u8, cwd: []const u8) void {
        // The in-process host's reaping rule, and its reason: the core reuses
        // pane ids and there is no close effect, so a deleted pane's shell
        // lives in its slot until a respawn lands here.
        if (a.ptys[pane]) |*old| {
            old.reader.cancel(a.io) catch {};
            _ = libc.close(old.file.handle);
            a.ptys[pane] = null;
        }
        a.gens[pane] +%= 1;
        var cwd_buf: [256:0]u8 = undefined;
        var cwd_z: ?[*:0]const u8 = null;
        if (cwd.len > 0 and cwd.len < cwd_buf.len) {
            @memcpy(cwd_buf[0..cwd.len], cwd);
            cwd_buf[cwd.len] = 0;
            cwd_z = @ptrCast(&cwd_buf);
        }
        // The SESSION's grid, which is what its panes are laid out against; the
        // pane's own size follows immediately as a `pty_resize`.
        //
        // `config.default_shell` and not the session's configured one: `Shell
        // <bin>` is a core setting, no message carries it, and inventing a
        // second place that decides which shell runs would be worse than one
        // that is occasionally the default. `shell_bin.resolve` falls back from
        // there exactly as it does for a whole session.
        const child = forkShell(null, pane, a.prompt_rcs, config.default_shell, cwd_z, a.client.rows, a.client.cols, null);
        a.ptys[pane] = .{ .file = child.file, .pid = child.pid, .reader = .{ .any_future = null, .result = {} } };
        // Unconditional, unlike `Shell.spawn`'s `threads_ok`: every spawn here
        // arrives over a socket this loop is already running, so there is no
        // pre-loop drain to be in.
        if (a.ptys[pane]) |*pt| {
            pt.reader = a.io.concurrent(readPty, .{ a.io, a.gpa, pt.file, @as(usize, pane), a.gens[pane], a.loop }) catch pt.reader;
        }
    }

    fn writeFile(a: *Attach, pane: u8, path: []const u8, bytes: []const u8) void {
        if (!writeFileBytes(path, bytes)) return;
        // Our own write is about to come back as a watch event: restamp from
        // the bytes we just put there so it reads as "no change". Only when
        // this IS the path this slot is watching — a `Save <elsewhere>` must
        // not silence a real change to the file the pane has open. Same rule as
        // `Shell.writeFile`; the comparison is against the path the session
        // asked us to watch, because there is no pane here to ask.
        //
        // The "saved <path>" message that host also writes is a pane's message
        // row, which belongs to the core: a detached save is silent.
        const watched = a.watch_paths[pane] orelse return;
        if (!std.mem.eql(u8, watched, path)) return;
        if (a.watches[pane]) |*w| w.generation = .{ .text = std.hash.Wyhash.hash(0, bytes) };
    }

    fn watchFile(a: *Attach, pane: u8, path: []const u8, on: bool) void {
        if (a.watch_paths[pane]) |old| a.gpa.free(old);
        a.watch_paths[pane] = null;
        if (!on or path.len == 0) return file_watch.watchPane(a.inotify_fd, &a.watches, pane, null, 0, .{ .text = 0 });
        const owned = a.gpa.dupe(u8, path) catch return;
        // Seeded from what is on disk RIGHT NOW, so the first `file_changed`
        // this sends is the first edit that is not already in the core. The
        // in-process host takes the same hash off the pane's live text; that
        // text is a socket away, and the file it came from is not.
        var hash: u64 = 0;
        if (look.readFile(a.gpa, owned)) |bytes| {
            hash = std.hash.Wyhash.hash(0, bytes);
            a.gpa.free(bytes);
        } else |_| {}
        a.watch_paths[pane] = owned;
        file_watch.watchPane(a.inotify_fd, &a.watches, pane, owned, 0, .{ .text = hash });
    }

    /// A coalesced inotify wake: re-read every watched path and hand the
    /// session the ones that really changed. It sends BYTES rather than
    /// reloading anything, because the text belongs to the core — which is
    /// exactly why `ClientTag` has a `file_changed` at all.
    fn reloadWatched(a: *Attach) ?AttachEnd {
        if (!a.check_files) return null;
        a.check_files = false;
        for (a.watch_paths, 0..) |slot, pane| {
            const path = slot orelse continue;
            const w = if (a.watches[pane]) |*entry| entry else continue;
            const bytes = look.readFile(a.gpa, path) catch continue;
            defer a.gpa.free(bytes);
            const hash = std.hash.Wyhash.hash(0, bytes);
            switch (w.generation) {
                .text => |accepted| if (accepted == hash) continue,
                // Never stored by this frontend: it cannot tell a PDF pane from
                // a text one (the message carries a path and nothing else), so
                // every slot is hashed and the core decides what the bytes mean
                // — `applyWatchedFileChanged` reopens the path for a PDF pane
                // and ignores them.
                .pdf => {},
            }
            // Committed here rather than after an acknowledgement, because
            // there is none: `file_changed` is a one-way event like every other
            // input on this wire. A snapshot the core rejects is therefore not
            // retried until the file changes again — the same bound the
            // in-process host lives with whenever a reload fails.
            w.generation = .{ .text = hash };
            if (a.send(.{ .file_changed = .{ .pane = @intCast(pane), .bytes = bytes } })) |end| return end;
        }
        return null;
    }

    /// The themes land on this frontend's disk. Where they went is reported on
    /// a pane's message row by the in-process host, and that row is the core's,
    /// so a detached dump is silent — the same shape as `write_dump` above.
    fn dumpThemes(a: *Attach) void {
        const dir = a.config_dir orelse return;
        const out = user_config.dumpThemes(a.io, a.gpa, dir, pardes.themes) catch return;
        a.gpa.free(out);
    }

    /// The capability handshake, resolved on the loop exactly as
    /// `Shell.pollFrame` resolves it and for its reason: the replies land on
    /// vaxis's reader thread, and this is the only thread allowed to write to
    /// the tty. No `native_images` here — this wire carries no attachments.
    fn enableCaps(a: *Attach) void {
        if (!a.caps_pending or !a.vx.queries_done.load(.unordered)) return;
        a.caps_pending = false;
        a.vx.enableDetectedFeatures(a.tty.writer()) catch {};
        // Earlier frames were drawn under the pre-handshake caps, and vaxis's
        // shadow grid has to be re-established under the new ones or it keeps
        // skipping cells it thinks are current.
        a.vx.queueRefresh();
        a.dirty = true;
    }

    fn paint(a: *Attach) void {
        a.dirty = false;
        const win = a.vx.window();
        // The session grid can be smaller than this window; `paintCells` clears
        // first, so the surplus is the terminal's own default cell rather than
        // whatever was there a frame ago.
        paintCells(win, a.client.grid.items, a.client.cols, a.client.rows);
        if (a.client.cursor) |cur| paintCursor(win, cur.x, cur.y, cur.bar);
        a.vx.render(a.tty.writer()) catch {};
    }
};

/// How long the frontend may sleep on the socket before it looks at the
/// terminal. This loop has TWO event sources and can block on only one of
/// them: vaxis delivers the terminal's events on its reader thread into a
/// mutex/condvar queue, which has no descriptor to hand `poll(2)` alongside
/// the socket — client.zig's `wait` takes a timeout for exactly that reason
/// ("the terminal it draws on is polled by whoever owns that"), and this is
/// whoever.
///
/// 8 ms is half a 60 Hz frame: a keystroke waits at most one of those before it
/// is on the wire (4 ms on average), and the frame it causes needs no wait at
/// all — it lands in the poll the moment the session writes it. The price of
/// the ceiling is 125 poll rounds a second on a frontend nobody is touching,
/// and it is measurably below the noise of what an idle pardes already costs:
/// on this machine (i7-11700, 100 Hz jiffies) an idle attached frontend used
/// 0.16% of one core over 60 s and 0.18% over 120 s, against 0.11% and 0.31%
/// for an idle in-process session on the same screen over the same windows.
/// Reach for an eventfd and a waker thread — fuse.zig's `pollLoop` is the
/// pattern — only if that ever stops being true.
const attach_poll_ms = 8;

/// The whole `--attach` session: connect, then one loop over the two event
/// sources until the session, the link or the terminal ends it. The terminal is
/// already raw, on the alt screen and reporting the mouse — `run` did that, and
/// `run`'s defers undo it, which is what makes an attach leave a terminal in
/// exactly the state an ordinary exit does.
fn attachSession(init: std.process.Init, opts: pardes.Options, name: []const u8, tty: *vaxis.Tty, vx: *vaxis.Vaxis) AttachEnd {
    const io = init.io;
    const gpa = init.gpa;

    // The hello carries this window, so the size has to be real before it goes
    // out: a session told 80x24 by a 200x50 terminal reflows every pane twice,
    // once now and once on the first SIGWINCH. `vx.resize` here rather than
    // waiting for the loop's first event for the same reason — the first frame
    // may arrive before any terminal event does, and it has to have somewhere
    // to be painted.
    const ws = tty.getWinsize() catch |err| return .{ .lost = err };
    if (ws.cols == 0 or ws.rows == 0) return .{ .lost = error.NoWinsize };
    vx.resize(gpa, tty.writer(), ws) catch |err| return .{ .lost = err };

    var client = detached_client.Client.open(gpa, name, ws.cols, ws.rows) catch |err| return switch (err) {
        error.NoSession, error.NoSessionPath => .{ .no_session = name },
        // The connect landed and the session hung up during the hello. That is
        // what a refusal AT ACCEPT TIME looks like from here: server.zig's
        // `refuseFd` writes six bytes and closes in the same pass, so the close
        // can beat our hello onto the socket and `open` never gets far enough
        // to read the reason. A refusal we do read arrives as `.refused` with
        // the reason in it.
        error.Closed => .rejected,
        else => .{ .lost = err },
    };
    // `detach` and not `deinit`: seven bytes that turn "the peer vanished" into
    // "the peer left" in the session's log.
    defer client.detach();

    var loop: Loop = .init(io, tty, vx);
    var paste_buf: std.Io.Writer.Allocating = .init(gpa);
    defer paste_buf.deinit();
    // Private and complete before any fork, exactly as in `run`: the pane
    // shells this frontend is asked to spawn borrow these stable path buffers.
    var prompt_rcs = shell_bin.PromptRcs.init();
    defer prompt_rcs.deinit();

    var a: Attach = .{
        .io = io,
        .gpa = gpa,
        .client = &client,
        .loop = &loop,
        .vx = vx,
        .tty = tty,
        .prompt_rcs = &prompt_rcs,
        .paste_buf = &paste_buf,
        .config_dir = opts.config_dir,
        .inotify_fd = if (builtin.os.tag == .linux) libc.inotify_init1(linux.IN.CLOEXEC) else -1,
    };
    // Registered BEFORE `loop.stop()` below so LIFO runs it after: the reader
    // has to be joined before the queue is emptied, or a late post lands in a
    // queue nobody drains again and its bytes leak. `run`'s teardown has the
    // same shape and the same order, minus the workers this frontend never
    // starts.
    defer {
        for (&a.ptys) |*slot| if (slot.*) |*pt| {
            pt.reader.cancel(io) catch {};
            _ = libc.close(pt.file.handle);
            slot.* = null;
        };
        if (a.watch_task) |*t| {
            t.cancel(io) catch {};
            a.watch_task = null;
        }
        if (a.inotify_fd >= 0) {
            _ = libc.close(a.inotify_fd);
            a.inotify_fd = -1;
        }
        for (&a.watch_paths) |*slot| if (slot.*) |p| {
            gpa.free(p);
            slot.* = null;
        };
        while (loop.tryEvent() catch null) |ev| switch (ev) {
            .pty_read => |pr| gpa.free(pr.bytes),
            .command => |line| gpa.free(line),
            .paste => |b| gpa.free(@constCast(b)),
            else => {},
        };
    }

    // A pardes started inside one of THIS frontend's pane shells finds this
    // process as its outer instance — the shells are our children — so the
    // nested-instance listener belongs here and not in the session, which has
    // no children at all. The command line it accepts goes over the wire as an
    // `Event.command`, which is what `ClientTag.command` is for.
    const sock_fd: c_int = if (opts.nested) -1 else nested.listen();
    defer nested.unlisten(sock_fd);

    loop.start() catch |err| return .{ .lost = err };
    defer loop.stop();
    // Both detached threads, for `run`'s reasons: sigwait and accept4 never
    // return, so an `io.concurrent` task around either would hang the teardown
    // that joins it.
    (std.Thread.spawn(.{}, winchWatch, .{ &loop, vx, tty }) catch |err| return .{ .lost = err }).detach();
    if (sock_fd >= 0) (std.Thread.spawn(.{}, lookServer, .{ gpa, sock_fd, &loop }) catch |err| return .{ .lost = err }).detach();
    if (a.inotify_fd >= 0) a.watch_task = io.concurrent(watchFiles, .{ io, a.inotify_fd, &loop }) catch null;
    // Send the capability probes and do not wait on them; `Attach.enableCaps`
    // resolves them on the loop. run() has the long version of why.
    vx.queryTerminalSend(tty.writer()) catch {};

    while (true) {
        const link = a.client.wait(attach_poll_ms);
        // DECODE BEFORE REACTING TO THE HANGUP. `wait` reports the close in
        // the same call that read the last bytes, and the last bytes are the
        // session's `quit`: `fill` appends every chunk and only then sees the
        // zero-length read. client.zig prefers POLLIN over POLLHUP for exactly
        // this reason, and honouring that means draining what arrived before
        // deciding the link is what ended us — otherwise an ordinary `Kill`
        // exits one frontend 0 (it got the quit alone) and whichever frontend
        // was in the same poll round nonzero, which is what the first run of
        // this loop actually did.
        while (true) {
            const msg = (a.client.next() catch |err| return .{ .lost = err }) orelse break;
            if (a.handle(msg)) |end| return end;
        }
        link catch |err| return .{ .lost = err };
        // The terminal, drained the way `Shell.waitInput` drains it and for its
        // reason: a wheel flick is one batch rather than fifty round trips, and
        // a paste in flight keeps draining without a message per character.
        var batch: usize = 0;
        while (a.in_paste or batch < 64) {
            // Propagated rather than swallowed, unlike `Shell.waitInput`'s
            // identical drain: there the blocking `nextEvent` above it is what
            // notices a dead event source, and here there is no blocking read
            // to notice with.
            const ev = (loop.tryEvent() catch |err| return .{ .lost = err }) orelse break;
            batch += 1;
            if (a.apply(ev)) |end| return end;
        }
        if (a.reloadWatched()) |end| return end;
        a.enableCaps();
        if (a.dirty) a.paint();
    }
}